35 lines
837 B
Python
35 lines
837 B
Python
from collections.abc import Iterable
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.exc import IntegrityError
|
|
from sqlalchemy.orm import declarative_base
|
|
from typing import Iterable, Self
|
|
|
|
Base = declarative_base()
|
|
Base.db = None
|
|
|
|
|
|
async def base_add(cls, stops: Self | Iterable[Self]) -> bool:
|
|
try:
|
|
method = (
|
|
cls.db.session.add_all
|
|
if isinstance(stops, Iterable)
|
|
else cls.db.session.add
|
|
)
|
|
method(stops)
|
|
await cls.db.session.commit()
|
|
except IntegrityError as err:
|
|
print(err)
|
|
|
|
|
|
Base.add = classmethod(base_add)
|
|
|
|
|
|
async def base_get_by_id(cls, id_: int | str) -> None | Base:
|
|
res = await cls.db.session.execute(select(cls).where(cls.id == id_))
|
|
element = res.scalar_one_or_none()
|
|
return element
|
|
|
|
|
|
Base.get_by_id = classmethod(base_get_by_id)
|