Bruno Charest
8affa0f8b7
Some checks failed
Tests / Backend Tests (Python) (3.10) (push) Has been cancelled
Tests / Backend Tests (Python) (3.11) (push) Has been cancelled
Tests / Backend Tests (Python) (3.12) (push) Has been cancelled
Tests / Frontend Tests (JS) (push) Has been cancelled
Tests / Integration Tests (push) Has been cancelled
Tests / All Tests Passed (push) Has been cancelled
72 lines
2.5 KiB
Python
72 lines
2.5 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import Optional
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models.container_customization import ContainerCustomization
|
|
|
|
|
|
class ContainerCustomizationRepository:
|
|
def __init__(self, session: AsyncSession):
|
|
self.session = session
|
|
|
|
async def list_by_user(self, user_id: Optional[int]) -> list[ContainerCustomization]:
|
|
stmt = select(ContainerCustomization)
|
|
if user_id is None:
|
|
stmt = stmt.where(ContainerCustomization.user_id.is_(None))
|
|
else:
|
|
stmt = stmt.where(ContainerCustomization.user_id == user_id)
|
|
stmt = stmt.order_by(ContainerCustomization.host_id.asc(), ContainerCustomization.container_id.asc())
|
|
result = await self.session.execute(stmt)
|
|
return list(result.scalars().all())
|
|
|
|
async def get_for_user(
|
|
self, *, user_id: Optional[int], host_id: str, container_id: str
|
|
) -> Optional[ContainerCustomization]:
|
|
stmt = select(ContainerCustomization).where(
|
|
ContainerCustomization.host_id == host_id,
|
|
ContainerCustomization.container_id == container_id,
|
|
)
|
|
if user_id is None:
|
|
stmt = stmt.where(ContainerCustomization.user_id.is_(None))
|
|
else:
|
|
stmt = stmt.where(ContainerCustomization.user_id == user_id)
|
|
result = await self.session.execute(stmt.limit(1))
|
|
return result.scalar_one_or_none()
|
|
|
|
async def upsert(
|
|
self,
|
|
*,
|
|
user_id: Optional[int],
|
|
host_id: str,
|
|
container_id: str,
|
|
icon_key: Optional[str],
|
|
icon_color: Optional[str],
|
|
bg_color: Optional[str],
|
|
) -> ContainerCustomization:
|
|
existing = await self.get_for_user(user_id=user_id, host_id=host_id, container_id=container_id)
|
|
if existing:
|
|
existing.icon_key = icon_key
|
|
existing.icon_color = icon_color
|
|
existing.bg_color = bg_color
|
|
await self.session.flush()
|
|
return existing
|
|
|
|
item = ContainerCustomization(
|
|
user_id=user_id,
|
|
host_id=host_id,
|
|
container_id=container_id,
|
|
icon_key=icon_key,
|
|
icon_color=icon_color,
|
|
bg_color=bg_color,
|
|
)
|
|
self.session.add(item)
|
|
await self.session.flush()
|
|
return item
|
|
|
|
async def delete(self, item: ContainerCustomization) -> None:
|
|
await self.session.delete(item)
|
|
await self.session.flush()
|