81 lines
2.3 KiB
Python
81 lines
2.3 KiB
Python
from typing import List
|
|
from sqlalchemy import Column, String
|
|
from sqlalchemy import update as sqlalchemy_update, delete as sqlalchemy_delete
|
|
from sqlalchemy.future import select
|
|
|
|
from app.database.db import Base, async_session
|
|
|
|
from uuid import uuid4
|
|
from typing_extensions import Self
|
|
|
|
class BaseCRUD(Base):
|
|
__abstract__ = True
|
|
id = Column(String, primary_key=True)
|
|
|
|
def __repr__(self):
|
|
return (
|
|
f"<{self.__class__.__name__}("
|
|
f"id={self.id}, "
|
|
f"name={self.name}, "
|
|
f")>"
|
|
)
|
|
|
|
@classmethod
|
|
async def create(cls, **kwargs) -> Self:
|
|
async with async_session() as db:
|
|
server = cls(id=str(uuid4()), **kwargs)
|
|
db.add(server)
|
|
try:
|
|
await db.commit()
|
|
await db.refresh(server)
|
|
except Exception:
|
|
await db.rollback()
|
|
raise
|
|
return server
|
|
|
|
@classmethod
|
|
async def update(cls, id, **kwargs) -> Self:
|
|
async with async_session() as db:
|
|
query = (
|
|
sqlalchemy_update(cls)
|
|
.where(cls.id == id)
|
|
.values(**kwargs)
|
|
.execution_options(synchronize_session="fetch")
|
|
)
|
|
await db.execute(query)
|
|
try:
|
|
await db.commit()
|
|
except Exception:
|
|
await db.rollback()
|
|
raise
|
|
return await cls.get(id)
|
|
|
|
@classmethod
|
|
async def get(cls, id) -> Self:
|
|
async with async_session() as db:
|
|
query = select(cls).where(cls.id == id)
|
|
servers = await db.execute(query)
|
|
(server,) = servers.first()
|
|
return server
|
|
|
|
@classmethod
|
|
async def get_all(cls) -> List[Self]:
|
|
async with async_session() as db:
|
|
query = select(cls)
|
|
servers = await db.execute(query)
|
|
servers = servers.scalars().all()
|
|
return servers
|
|
|
|
@classmethod
|
|
async def delete(cls, id) -> bool:
|
|
async with async_session() as db:
|
|
query = sqlalchemy_delete(cls).where(cls.id == id)
|
|
await db.execute(query)
|
|
try:
|
|
await db.commit()
|
|
except Exception:
|
|
await db.rollback()
|
|
raise
|
|
return True
|
|
|