- add marshmallow-sqlalchemy marshmallow for inserting db data via deserialization

This commit is contained in:
Arno Kaimbacher 2022-02-22 16:36:49 +01:00
parent c62eb2cdb0
commit 1dafc5824c
10 changed files with 118 additions and 189 deletions

15
.vscode/launch.json vendored Normal file
View File

@ -0,0 +1,15 @@
{
// Use IntelliSense to learn about possible attributes.
// Hover to view descriptions of existing attributes.
// For more information, visit: https://go.microsoft.com/fwlink/?linkid=830387
"version": "0.2.0",
"configurations": [
{
"name": "Python: Current File",
"type": "python",
"request": "launch",
"program": "${file}",
"console": "integratedTerminal"
}
]
}

View File

@ -7,6 +7,7 @@
"pylint_flask",
"pylint_flask_sqlalchemy",
],
"python.pythonPath": "d:\\Software\\geomon\\.venv\\Scripts\\python.exe",
// "python.pythonPath": "d:\\Software\\geomon\\.venv\\Scripts\\python.exe",
"python.defaultInterpreterPath": "./.venv/Scripts/python.exe",
"python.envFile": "${workspaceFolder}/.env"
}

View File

@ -1,2 +1,3 @@
# For relative imports to work in Python 3.6
import os, sys; sys.path.append(os.path.dirname(os.path.realpath(__file__)))
import os, sys; sys.path.append(os.path.dirname(os.path.realpath(__file__)))
# print(f'Invoking __init__.py for {__name__}')

View File

@ -5,8 +5,8 @@ from typing import List
import uuid
from sqlalchemy.orm import session
from sqlalchemy import func, desc, asc
from fb_models import (create_session, FbObservation, Catena)
from pg_models import (create_pg_session, Dataset, Observation, Procedure, Phenomenon, Platform)
from db.fb_models import (create_session, FbObservation, Catena)
from db.pg_models import (create_pg_session, Dataset, Observation, Procedure, Phenomenon, Platform)
def main():
"""

View File

@ -1,15 +1,82 @@
'''
Tutorial link: https://realpython.com/flask-connexion-rest-api-part-2/
https://github.com/realpython/materials/blob/master/flask-connexion-rest-part-2/version_1/people.py
Sqlalchemy version: 1.2.15
Python version: 3.7
'''
import os
# import sys, inspect
# currentdir = os.path.dirname(os.path.abspath(inspect.getfile(inspect.currentframe())))
# parentdir = os.path.dirname(currentdir)
# sys.path.insert(0, parentdir)
from db.pg_models import create_pg_session
from sqlalchemy.orm import session
from gschliefgraben_glasfaser.models import PersonSchema
from models import Person, PersonSchema
# response = requests.get('https://api.com/')
# print(response) # shows the response's HTTP status code
# print(response.json()) # shows the response's JSON response body, if it has one
# print(response.content) # get the data content of the response
# (for your case this is the downloaded file)
# print(dir(response)) # shows you all the different methods you can call on this response object
db_user = os.environ.get("POSTGIS_DBUSER")
print(db_user)
def main():
# db_user = os.environ.get("POSTGIS_DBUSER")
# print(db_user)
pg_session: session = create_pg_session()
pg_person: Person = pg_session.query(Person).first()
# serialize db data to json
person_schema = PersonSchema()
dump_data = person_schema.dump(pg_person)
print(dump_data)
# deserialize
load_data: Person = person_schema.load(dump_data)
print(load_data)
create(dump_data)
def create(person_json: PersonSchema):
"""
This function creates a new person in the people structure
based on the passed-in person data
:param person: person to create in people structure
:return: 201 on success, 406 on person exists
"""
login = person_json.get('login')
#lname = person.get('lname')
session = create_pg_session()
# existing_person = Person.query \
# .filter(Person.login == login) \
# .one_or_none()
existing_person: bool = ( \
session.query(Person) \
.filter(Person.login == login)
.one_or_none()
)
# Can we insert this person?
if existing_person is None:
# Create a person instance using the schema and the passed in person
schema = PersonSchema()
# deserialize to object
new_person: Person = schema.load(person_json)
# Add the person to the database
session.add(new_person)
session.commit()
# Serialize and return the newly created person in the response
data = schema.dump(new_person)
return data, 201
# Otherwise, nope, person exists already
else:
print(409, f'Person {login} exists already')
if __name__ == "__main__":
main()

View File

@ -1,8 +1,9 @@
'''
Tutorial link: https://docs.sqlalchemy.org/en/latest/orm/tutorial.html
Sqlalchemy version: 1.2.15
Python version: 3.7
Sqlalchemy version: 1.4.31
Python version: 3.10
'''
#!/usr/bin/python# -*- coding: utf-8 -*-
from datetime import datetime
# from config import db, ma
@ -14,24 +15,32 @@ from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import session
from marshmallow import Schema
from db.pg_models import create_pg_session
from marshmallow_sqlalchemy import SQLAlchemyAutoSchema
Base = declarative_base()
class Person(Base):
""" Platform class """
__tablename__ = 'person'
person_id = Column(Integer, primary_key=True)
lname = Column(String(32), index=True)
fname = Column(String(32))
timestamp = Column(DateTime, default=datetime.utcnow,
__tablename__ = 'accounts'
__table_args__ = {"schema": "gba"}
person_id = Column('id', Integer, primary_key=True)
lname = Column('last_name', String(255), index=True)
fname = Column('first_name', String(255))
login = Column(String(255))
timestamp = Column('updated_at', DateTime, default=datetime.utcnow,
onupdate=datetime.utcnow)
def __repr__(self):
return "<User(name='%s', lastname='%s')>" % (
self.login, self.lname)
class PersonSchema(Schema):
class PersonSchema(SQLAlchemyAutoSchema):
""" Platform class """
class Meta:
""" Platform class """
model = Person
include_relationships = True
load_instance = True
#pg_session: session = create_pg_session()
sqla_session: session = create_pg_session()

View File

@ -28,9 +28,9 @@ Install python formatter:
d:/Software/geomon/.venv/Scripts/python.exe -m pip install -U autopep8
pip install fdb
pip install sqlalchemy-firebird
pip install psycopg2
python -m pip install fdb
python -m pip install sqlalchemy-firebird
python -m pip install psycopg2
Marshmallow provides functionality to serialize and deserialize Python objects as they flow out of and into our JSON-based REST API. Marshmallow converts Python class instances to objects that can be converted to JSON.
pip install marshmallow-sqlalchemy marshmallowll
python -m pip install marshmallow-sqlalchemy marshmallow

View File

@ -1,169 +0,0 @@
'''
Tutorial link: https://docs.sqlalchemy.org/en/latest/orm/tutorial.html
Sqlalchemy version: 1.2.15
Python version: 3.7
'''
import os
from sqlalchemy import (create_engine, Column, Integer,
SmallInteger, String, ForeignKey, DateTime, Numeric)
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, relationship
import sqlalchemy.orm.session
Base = declarative_base()
class Platform(Base):
""" Platform class """
__tablename__ = 'platform'
__table_args__ = {"schema": "gba"}
id = Column('platform_id', Integer, primary_key=True)
identifier = Column('identifier', String)
sta_identifier = Column('sta_identifier', String)
name = Column('name', String)
# datasets = relationship('Dataset')
datasets = relationship('Dataset', back_populates="platform", lazy=True)
def __repr__(self):
return f'Platform {self.name}'
class Phenomenon(Base):
""" phenomenon class """
__tablename__ = 'phenomenon'
__table_args__ = {"schema": "gba"}
id = Column('phenomenon_id', Integer, primary_key=True)
name = Column('name', String)
sta_identifier = Column('sta_identifier', String)
# datasets = relationship('Dataset')
datasets = relationship('Dataset', back_populates="phenomenon", lazy=True)
def __repr__(self):
return f'Phenomenon {self.name}'
class Procedure(Base):
""" procedure class """
__tablename__ = 'procedure'
__table_args__ = {"schema": "gba"}
id = Column('procedure_id', Integer, primary_key=True)
name = Column('name', String)
sta_identifier = Column('sta_identifier', String)
# datasets = relationship('Dataset')
datasets = relationship('Dataset', back_populates="procedure", lazy=True)
def __repr__(self):
return f'Procedure {self.name}'
class Dataset(Base):
""" dataset class """
__tablename__ = 'dataset'
__table_args__ = {"schema": "gba"}
id = Column('dataset_id', Integer, primary_key=True)
name = Column('name', String)
is_published = Column('is_published', SmallInteger)
is_hidden = Column('is_hidden', SmallInteger)
dataset_type = Column('dataset_type', String)
observation_type = Column('observation_type', String)
value_type = Column('value_type', String)
last_time = Column('last_time', DateTime)
last_value = Column('last_value', Numeric(20, 10))
fk_last_observation_id = Column(
'fk_last_observation_id',
Integer
)
# last_observation = relationship(
# "Observation", foreign_keys=[fk_last_observation_id])
first_time = Column('first_time', DateTime)
first_value = Column('first_value', Numeric(20, 10))
fk_first_observation_id = Column(
'fk_first_observation_id',
Integer
)
# first_observation = relationship("Observation", foreign_keys=[
# fk_first_observation_id])
observations = relationship(
'Observation', back_populates='dataset', lazy=True)
fk_procedure_id = Column('fk_procedure_id', Integer, ForeignKey(
'gba.procedure.procedure_id'), nullable=False)
# procedure = relationship("Procedure", lazy="joined")
procedure = relationship(
"Procedure", back_populates="datasets", lazy="joined")
fk_phenomenon_id = Column(
'fk_phenomenon_id', Integer, ForeignKey('gba.phenomenon.phenomenon_id'), nullable=False)
# phenomenon = relationship("Phenomenon", lazy="joined", foreign_keys=[fk_phenomenon_id])
phenomenon = relationship(
"Phenomenon", back_populates="datasets", lazy="joined")
# fk_platform_id = Column(
# 'fk_platform_id', Integer, ForeignKey('gba.platform.platform_id'), nullable=True)
# # platform = relationship("Platform", lazy="joined", foreign_keys=[fk_platform_id])
fk_platform_id = Column('fk_platform_id', Integer, ForeignKey(
'gba.platform.platform_id'), nullable=True)
platform = relationship(
"Platform", back_populates="datasets", lazy="joined")
def __repr__(self):
return f'Dataset {self.name}'
class Observation(Base):
""" observation class """
__tablename__ = 'observation'
__table_args__ = {"schema": "gba"}
id = Column('observation_id', Integer, primary_key=True)
name = Column('name', String)
value_type = Column('value_type', String)
# pitch = Column('PITCH', String)
# roll = Column('ROLL', String)
sampling_time_start = Column('sampling_time_start', DateTime)
sampling_time_end = Column('sampling_time_end', DateTime)
result_time = Column('result_time', DateTime)
sta_identifier = Column('sta_identifier', String)
value_quantity = Column('value_quantity', Numeric(20, 10), nullable=False)
# fk_dataset_id = Column('fk_dataset_id', Integer,
# ForeignKey('gba.dataset.dataset_id'))
# dataset = relationship("Dataset", lazy="joined",
# foreign_keys=[fk_dataset_id])
fk_dataset_id = Column(Integer, ForeignKey(
'gba.dataset.dataset_id'), nullable=False)
dataset = relationship("Dataset", back_populates="observations")
def __repr__(self):
return f'Observation {self.name}'
# @property
# def result_time(self):
# ''' Create a datetime object '''
# start_datetime = datetime.datetime.combine(self.date, self.ora)
# return start_datetime
def create_pg_session() -> sqlalchemy.orm.sessionmaker:
"""Return the sum of x and y."""
dbschema = ''
db_user = os.environ.get("POSTGIS_DBUSER")
db_password = os.environ.get("POSTGIS_DBPASSWORD")
db_url = os.environ.get("POSTGIS_DBURL")
engine = create_engine(
"postgresql+psycopg2://" + db_user + ":" + db_password + "@" + db_url,
connect_args={'options': '-csearch_path={}'.format(dbschema)},
isolation_level="READ UNCOMMITTED")
session_maker = sessionmaker(bind=engine)
session = session_maker()
Base.metadata.create_all(engine)
return session

5
setup.py Normal file
View File

@ -0,0 +1,5 @@
from setuptools import setup, find_packages
setup(
name = 'geomon',
packages = find_packages(),
)