python 一些代码使用模式作为SQL-Alchemy的声明基础，以及对Celery分支的支持。

Posted 2021-05-09

from sqlalchemy import create_engine, MetaData, Table
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base
from multiprocessing.util import register_after_fork


Base = declarative_base()


_ENGINES = {}
_SESSIONS = {}

DATABASE_URL='insert ur here'

# https://github.com/celery/celery/blob/master/celery/backends/database/session.py#L27-L46

class _after_fork(object):
    registered = False

    def __call__(self):
        self.registered = False  # child must reregister
        for engine in list(_ENGINES.values()):
            engine.dispose()
        _ENGINES.clear()
        _SESSIONS.clear()
after_fork = _after_fork()

def get_engine(dburi, **kwargs):
    try:
        return _ENGINES[dburi]
    except KeyError:
        engine = _ENGINES[dburi] = create_engine(dburi, **kwargs)
        after_fork.registered = True
        register_after_fork(after_fork, after_fork)
        return engine


def create_session(dburi, short_lived_sessions=False, **kwargs):
    engine = get_engine(dburi, **kwargs)
    if short_lived_sessions or dburi not in _SESSIONS:
        _SESSIONS[dburi] = sessionmaker(bind=engine)
    return engine, _SESSIONS[dburi]

engine = create_engine(DATABASE_URL)
metadata = MetaData(bind=engine)

session_maker = create_session(DATABASE_URL)[1]


class SessionPropertyMixin(object):
    def save(self, session=None, commit=True, close=False):
        if session:
            session.add(self)
            session.commit()
        if close:
            session.close()


# Reflect each database table we need to use, via schemas.
class Slugs(Base, SessionPropertyMixin):
    __table__ = Table('slugs', metadata, autoload=True)