67 lines
2.0 KiB
Python
67 lines
2.0 KiB
Python
'''
|
|
Tutorial link: https://docs.sqlalchemy.org/en/latest/orm/tutorial.html
|
|
Sqlalchemy version: 1.2.15
|
|
Python version: 3.7
|
|
'''
|
|
import os
|
|
import datetime
|
|
from sqlalchemy import (create_engine, Column, Integer,
|
|
String, ForeignKey, Time, Date)
|
|
from sqlalchemy.ext.declarative import declarative_base
|
|
from sqlalchemy.orm import sessionmaker, relationship
|
|
import sqlalchemy.orm.session
|
|
|
|
Base = declarative_base()
|
|
|
|
|
|
class Catena(Base):
|
|
""" catena class """
|
|
__tablename__ = 'CATENE'
|
|
|
|
id = Column('CHIAVE', Integer, primary_key=True)
|
|
name = Column('NOME', String)
|
|
observations = relationship('FbObservation')
|
|
|
|
def __repr__(self): # optional
|
|
return f'Catena {self.name}'
|
|
|
|
|
|
class FbObservation(Base):
|
|
""" FbObservation class """
|
|
__tablename__ = 'DATI_ACQUISITI'
|
|
|
|
# id = Column(Integer, primary_key=True) # obligatory
|
|
pitch = Column('PITCH', String)
|
|
roll = Column('ROLL', String)
|
|
ora = Column('ORA', Time, primary_key=True)
|
|
sensore = Column('SENSORE', Integer, primary_key=True)
|
|
data = Column('DATA', Date)
|
|
temperature = Column('TEMPERATURA', String)
|
|
|
|
chiave_id = Column('CATENA', Integer, ForeignKey('CATENE.CHIAVE'))
|
|
catena = relationship("Catena", lazy="joined", foreign_keys=[chiave_id])
|
|
|
|
def __repr__(self): # optional
|
|
return f'FbObservation {self.roll}'
|
|
|
|
@property
|
|
def result_time(self):
|
|
''' Create a datetime object '''
|
|
start_datetime = datetime.datetime.combine(self.data, self.ora)
|
|
return start_datetime
|
|
|
|
|
|
def create_session() -> sqlalchemy.orm.session:
|
|
"""Return the sum of x and y."""
|
|
# engine = create_engine('sqlite:///:memory:')
|
|
db_user = os.environ.get("FDB_DBUSER")
|
|
db_password = os.environ.get("FDB_DBPASSWORD")
|
|
db_url = os.environ.get("FDB_DBURL")
|
|
engine = create_engine(
|
|
"firebird+fdb://" + db_user + ":" + db_password + "@" + db_url)
|
|
session_maker = sessionmaker(bind=engine)
|
|
session = session_maker()
|
|
|
|
Base.metadata.create_all(engine)
|
|
return session
|