Bruno Charest
661d005fc7
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
129 lines
5.0 KiB
Python
129 lines
5.0 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import Optional
|
|
|
|
from sqlalchemy import delete, select, update
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models.favorite_container import FavoriteContainer
|
|
from app.models.favorite_group import FavoriteGroup
|
|
|
|
|
|
class FavoriteGroupRepository:
|
|
def __init__(self, session: AsyncSession):
|
|
self.session = session
|
|
|
|
async def list_by_user(self, user_id: Optional[int]) -> list[FavoriteGroup]:
|
|
stmt = select(FavoriteGroup)
|
|
if user_id is None:
|
|
stmt = stmt.where(FavoriteGroup.user_id.is_(None))
|
|
else:
|
|
stmt = stmt.where(FavoriteGroup.user_id == user_id)
|
|
stmt = stmt.order_by(FavoriteGroup.sort_order.asc(), FavoriteGroup.name.asc())
|
|
result = await self.session.execute(stmt)
|
|
return list(result.scalars().all())
|
|
|
|
async def get_for_user(self, group_id: int, user_id: Optional[int]) -> Optional[FavoriteGroup]:
|
|
stmt = select(FavoriteGroup).where(FavoriteGroup.id == group_id)
|
|
if user_id is None:
|
|
stmt = stmt.where(FavoriteGroup.user_id.is_(None))
|
|
else:
|
|
stmt = stmt.where(FavoriteGroup.user_id == user_id)
|
|
result = await self.session.execute(stmt)
|
|
return result.scalar_one_or_none()
|
|
|
|
async def exists_name(self, name: str, user_id: Optional[int]) -> bool:
|
|
stmt = select(FavoriteGroup.id).where(FavoriteGroup.name == name)
|
|
if user_id is None:
|
|
stmt = stmt.where(FavoriteGroup.user_id.is_(None))
|
|
else:
|
|
stmt = stmt.where(FavoriteGroup.user_id == user_id)
|
|
result = await self.session.execute(stmt.limit(1))
|
|
return result.scalar_one_or_none() is not None
|
|
|
|
async def create(
|
|
self,
|
|
*,
|
|
user_id: Optional[int],
|
|
name: str,
|
|
sort_order: int = 0,
|
|
color: Optional[str] = None,
|
|
icon_key: Optional[str] = None,
|
|
) -> FavoriteGroup:
|
|
group = FavoriteGroup(user_id=user_id, name=name, sort_order=sort_order, color=color, icon_key=icon_key)
|
|
self.session.add(group)
|
|
await self.session.flush()
|
|
return group
|
|
|
|
async def update(
|
|
self,
|
|
group: FavoriteGroup,
|
|
*,
|
|
name: Optional[str] = None,
|
|
sort_order: Optional[int] = None,
|
|
color: Optional[str] = None,
|
|
icon_key: Optional[str] = None,
|
|
) -> FavoriteGroup:
|
|
if name is not None:
|
|
group.name = name
|
|
if sort_order is not None:
|
|
group.sort_order = sort_order
|
|
if color is not None:
|
|
group.color = color
|
|
if icon_key is not None:
|
|
group.icon_key = icon_key
|
|
await self.session.flush()
|
|
return group
|
|
|
|
async def delete(self, group: FavoriteGroup) -> None:
|
|
await self.session.delete(group)
|
|
await self.session.flush()
|
|
|
|
|
|
class FavoriteContainerRepository:
|
|
def __init__(self, session: AsyncSession):
|
|
self.session = session
|
|
|
|
async def list_by_user(self, user_id: Optional[int]) -> list[FavoriteContainer]:
|
|
stmt = select(FavoriteContainer)
|
|
if user_id is None:
|
|
stmt = stmt.where(FavoriteContainer.user_id.is_(None))
|
|
else:
|
|
stmt = stmt.where(FavoriteContainer.user_id == user_id)
|
|
stmt = stmt.order_by(FavoriteContainer.created_at.desc())
|
|
result = await self.session.execute(stmt)
|
|
return list(result.scalars().all())
|
|
|
|
async def get_for_user(self, favorite_id: int, user_id: Optional[int]) -> Optional[FavoriteContainer]:
|
|
stmt = select(FavoriteContainer).where(FavoriteContainer.id == favorite_id)
|
|
if user_id is None:
|
|
stmt = stmt.where(FavoriteContainer.user_id.is_(None))
|
|
else:
|
|
stmt = stmt.where(FavoriteContainer.user_id == user_id)
|
|
result = await self.session.execute(stmt)
|
|
return result.scalar_one_or_none()
|
|
|
|
async def get_by_docker_container_id(self, docker_container_db_id: int, user_id: Optional[int]) -> Optional[FavoriteContainer]:
|
|
stmt = select(FavoriteContainer).where(FavoriteContainer.docker_container_id == docker_container_db_id)
|
|
if user_id is None:
|
|
stmt = stmt.where(FavoriteContainer.user_id.is_(None))
|
|
else:
|
|
stmt = stmt.where(FavoriteContainer.user_id == user_id)
|
|
result = await self.session.execute(stmt.limit(1))
|
|
return result.scalar_one_or_none()
|
|
|
|
async def create(self, *, user_id: Optional[int], docker_container_db_id: int, group_id: Optional[int]) -> FavoriteContainer:
|
|
fav = FavoriteContainer(user_id=user_id, docker_container_id=docker_container_db_id, group_id=group_id)
|
|
self.session.add(fav)
|
|
await self.session.flush()
|
|
return fav
|
|
|
|
async def update_group(self, fav: FavoriteContainer, *, group_id: Optional[int]) -> FavoriteContainer:
|
|
fav.group_id = group_id
|
|
await self.session.flush()
|
|
return fav
|
|
|
|
async def delete(self, fav: FavoriteContainer) -> None:
|
|
await self.session.delete(fav)
|
|
await self.session.flush()
|