Skip to content

Instantly share code, notes, and snippets.

@afonasev
Created June 9, 2020 14:01
Show Gist options
  • Save afonasev/2e3ad61c27d603fa8bd2e9c896ed0087 to your computer and use it in GitHub Desktop.
Save afonasev/2e3ad61c27d603fa8bd2e9c896ed0087 to your computer and use it in GitHub Desktop.
Sqlalchemy BaseModel
from sqlalchemy.ext.declarative import DeclarativeMeta, declarative_base, declared_attr
from sqlalchemy import MetaData
from sqlalchemy.engine import create_engine
from sqlalchemy.orm import Session
from sqlalchemy.orm import scoped_session as _scoped_session
from sqlalchemy.orm import sessionmaker as _sessionmaker
from .config import config
engine = create_engine(
config.URI,
pool_size=config.POOL_SIZE,
pool_recycle=config.POOL_RECYCLE,
max_overflow=config.MAX_OVERFLOW,
connect_args=config.CONNECT_ARGS,
echo=config.ECHO,
)
meta = MetaData()
session_maker = _sessionmaker(
bind=engine,
autoflush=config.AUTOFLUSH,
expire_on_commit=config.EXPIRE_ON_COMMIT,
)
scoped_session = _scoped_session(session_maker)
class ModelBase:
query_class = None
query = scoped_session.query_property()
@declared_attr
def __tablename__(cls) -> str:
return cls.__name__.lower()
Model = declarative_base(
bind=engine,
cls=ModelBase,
metaclass=DeclarativeMeta,
metadata=meta,
)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment