62 lines
1.4 KiB
Python
62 lines
1.4 KiB
Python
from sqlalchemy import create_engine, Table
|
|
from sqlalchemy.ext.declarative import declarative_base
|
|
from sqlalchemy.orm import sessionmaker
|
|
from typing import Optional, Dict
|
|
from dataclasses import dataclass
|
|
import json
|
|
|
|
|
|
@dataclass
|
|
class Database:
|
|
engine: str
|
|
name: str
|
|
host: str
|
|
user: str
|
|
password: str
|
|
options: Optional[Dict] = None
|
|
|
|
def get_url(self):
|
|
option_str = ""
|
|
if self.options:
|
|
opt = [name+'='+'+'.join(value.split())
|
|
for name, value in self.options.items()]
|
|
option_str = "?"+"&".join(opt)
|
|
return f"{self.engine}://{self.user}:{self.password}@{self.host}/{self.name}{option_str}"
|
|
|
|
|
|
data = {}
|
|
|
|
with open('settings', 'r') as settings:
|
|
data_str = settings.read()
|
|
data = json.loads(data_str)
|
|
db = Database(**data.get('database'))
|
|
|
|
SQL_ALCHEMY_DATABASE_URL_MSSQL = db.get_url()
|
|
|
|
|
|
engine = create_engine(
|
|
SQL_ALCHEMY_DATABASE_URL_MSSQL, connect_args={"check_same_thread": False}
|
|
)
|
|
|
|
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
|
|
|
|
Base = declarative_base()
|
|
|
|
|
|
def get_table(tablename: str, dbname: str = None) -> Table:
|
|
metadata = Base.metadata
|
|
metadata.reflect(bind=engine)
|
|
|
|
if dbname == None:
|
|
dbname = db.name
|
|
|
|
return Table(tablename, metadata, schema=dbname+'.dbo', autoload=True, autoload_with=engine)
|
|
|
|
|
|
def get_db():
|
|
db = SessionLocal()
|
|
try:
|
|
yield db
|
|
finally:
|
|
db.close()
|