В моем модуле я определил base.py:
< pre class="lang-py Prettyprint-override">
Код: Выделить всё
from sqlalchemy import MetaData
from sqlalchemy.orm import DeclarativeBase, Session
class Base(DeclarativeBase):
# Ensure all models use the public schema by default
metadata = MetaData(schema='public')
__abstract__ = True
def __eq__(self, other):
return True
def to_dict(self) -> dict:
"""Converts the fields (columns) of a class to a dictionary without the '_sa_instance_state' key."""
return {field.name: getattr(self, field.name) for field in self.__table__.c}
@classmethod
def get_count(cls, session: Session) -> int:
"""Get the count of rows in the table."""
return session.query(cls).count()
Код: Выделить всё
from sqlalchemy import Column, DateTime, Integer, Float, ForeignKey
from sqlalchemy.orm import relationship
from .base import Base
class SensorData(Base):
__tablename__ = 'sensor_data'
trip_id = Column(Integer, primary_key=True)
timestamp = Column(DateTime, primary_key=True)
acceleration_x = Column(Float)
# Other fields...
Код: Выделить всё
from sqlalchemy import create_engine, MetaData
from .models import SensorData # Models imported after Base
from .base import Base
class Database:
def __init__(self):
from sqlalchemy.engine.url import URL
url_object = URL.create(
drivername="postgresql+psycopg2",
username="username",
password="password",
host="localhost",
port=5432,
database="database"
)
self.engine = create_engine(url_object, echo=True)
self.metadata = MetaData(schema='public')
# Explicitly bind metadata and engine to the Base class
Base.metadata = self.metadata
Base.metadata.bind = self.engine
Base.metadata.create_all(self.engine) # Table creation
Код: Выделить всё
db = Database()
Подробнее здесь: https://stackoverflow.com/questions/792 ... base-class