SQLAlchemy
الحالة::مؤرشفة
المراجع::


# base
from sqlalchemy.orm import declarative_base
 
ContentBase = declarative_base()
BroadcastBase = declarative_base()
 
# session
from src.db.base import BroadcastBase, ContentBase
 
content_db_engine = create_engine(
    f"sqlite:///{CONTENT_DB_PATH}",
    connect_args={"check_same_thread": False},
    json_serializer=lambda obj: json.dumps(obj, ensure_ascii=False),
)
broadcast_db_engine = create_engine(
    f"sqlite:///{BROADCAST_DB_PATH}", connect_args={"check_same_thread": False}
)
 
ContentBase.metadata.create_all(bind=content_db_engine)
BroadcastBase.metadata.create_all(bind=broadcast_db_engine)
 
session = scoped_session(
    sessionmaker(
        autocommit=False,
        autoflush=False,
        binds={
            ContentBase: content_db_engine,
            BroadcastBase: broadcast_db_engine,
        },
    )
)
 
# model
from src.db.base import BroadcastBase
 
class Broadcast(BroadcastBase):