''' 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') observations = relationship('FbObservation', back_populates="catena", lazy=True) 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, primary_key=True) temperature = Column('TEMPERATURA', String) chiave_id = Column('CATENA', Integer, ForeignKey('CATENE.CHIAVE'), primary_key=True, nullable=True) catena = relationship( "Catena", back_populates="observations", lazy="joined") # 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