You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
37 lines
1.0 KiB
Python
37 lines
1.0 KiB
Python
from __future__ import annotations
|
|
from typing import TypeVar
|
|
import sqlalchemy
|
|
from sqlalchemy.orm import DeclarativeBase
|
|
from sqlalchemy.ext.asyncio import AsyncAttrs
|
|
|
|
from service.database import DatabaseService
|
|
|
|
class BaseModel(AsyncAttrs, DeclarativeBase):
|
|
pass
|
|
|
|
class BaseHelper:
|
|
def __init__(self, dbs: DatabaseService):
|
|
self.dbs = dbs
|
|
self.initialized = False
|
|
|
|
async def __aenter__(self):
|
|
if not self.initialized:
|
|
self.create_session = self.dbs.create_session
|
|
self.session = self.dbs.create_session()
|
|
await self.session.__aenter__()
|
|
self.initialized = True
|
|
|
|
return self
|
|
|
|
async def __aexit__(self, exc_type, exc, tb):
|
|
await self.session.__aexit__(exc_type, exc, tb)
|
|
|
|
T = TypeVar("T", bound=BaseModel)
|
|
|
|
def clone_model(model: T) -> T:
|
|
data_dict = {}
|
|
for c in sqlalchemy.inspect(model).mapper.column_attrs:
|
|
if c.key == "id":
|
|
continue
|
|
data_dict[c.key] = getattr(model, c.key)
|
|
return model.__class__(**data_dict) |