ciclos

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

from sqlalchemy import create_engine, event
from sqlalchemy.orm import sessionmaker, scoped_session
from sqlalchemy.ext.declarative import declarative_base

from datetime import datetime

from .config import read_from_config

engine = create_engine(read_from_config('connection-string'), echo=True)
db = scoped_session(sessionmaker(autocommit=False, autoflush=False, bind=engine))

Base = declarative_base()
Base.query = db.query_property()

def init_database():
    @event.listens_for(Base, 'before_insert', propagate=True)
    def before_insert(mapper, connection, target):
        if hasattr(target, '_no_autoupdate'):
            return
        if hasattr(target, 'created_at'):
            target.created_at = datetime.utcnow()
        if hasattr(target, 'updated_at'):
            target.updated_at = datetime.utcnow()

    @event.listens_for(Base, 'before_update', propagate=True)
    def before_insert(mapper, connection, target):
        if hasattr(target, '_no_autoupdate'):
            return
        if hasattr(target, 'updated_at'):
            target.updated_at = datetime.utcnow()

    import core.types
    Base.metadata.create_all(bind=engine)