1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
|
from sqlalchemy import create_engine, engine
from sqlalchemy.orm import sessionmaker, Session
from typing import Dict, Any
from models import Base
class DbEngine:
def __init__(self, connection_str: str, **kwargs: Dict[str, Any]) -> None:
self.connection_str = connection_str
self.kwargs = kwargs
self.engine: engine = None
self.session_factory: sessionmaker = None
def initialize(self) -> None:
self.engine = create_engine(self.connection_str, **self.kwargs)
Base.metadata.bind = self.engine
self.session_factory = sessionmaker(bind=self.engine)
Base.metadata.create_all()
def session(self) -> Session:
return self.session_factory()
|