Daniil
64 lines
2.1 KiB
Python
64 lines
2.1 KiB
Python
from __future__ import annotations
|
|
|
|
import uuid
|
|
|
|
from sqlalchemy import Select, select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from cpv3.modules.projects.models import Project
|
|
from cpv3.modules.projects.schemas import ProjectCreate, ProjectUpdate
|
|
from cpv3.modules.users.models import User
|
|
|
|
|
|
class ProjectRepository:
|
|
"""Repository for Project database operations."""
|
|
|
|
def __init__(self, session: AsyncSession) -> None:
|
|
self._session = session
|
|
|
|
async def list_all(self, *, requester: User) -> list[Project]:
|
|
stmt: Select[tuple[Project]] = select(Project).where(
|
|
Project.is_active.is_(True)
|
|
)
|
|
if not requester.is_staff:
|
|
stmt = stmt.where(Project.owner_id == requester.id)
|
|
|
|
result = await self._session.execute(stmt.order_by(Project.created_at.desc()))
|
|
return list(result.scalars().all())
|
|
|
|
async def get_by_id(self, project_id: uuid.UUID) -> Project | None:
|
|
result = await self._session.execute(
|
|
select(Project)
|
|
.where(Project.id == project_id)
|
|
.where(Project.is_active.is_(True))
|
|
)
|
|
return result.scalar_one_or_none()
|
|
|
|
async def create(self, *, requester: User, data: ProjectCreate) -> Project:
|
|
project = Project(
|
|
owner_id=requester.id,
|
|
name=data.name,
|
|
description=data.description,
|
|
language=data.language,
|
|
folder=data.folder,
|
|
status=data.status,
|
|
)
|
|
|
|
self._session.add(project)
|
|
await self._session.commit()
|
|
await self._session.refresh(project)
|
|
return project
|
|
|
|
async def update(self, project: Project, data: ProjectUpdate) -> Project:
|
|
for key, value in data.model_dump(exclude_unset=True).items():
|
|
if value is not None:
|
|
setattr(project, key, value)
|
|
|
|
await self._session.commit()
|
|
await self._session.refresh(project)
|
|
return project
|
|
|
|
async def deactivate(self, project: Project) -> None:
|
|
project.is_active = False
|
|
await self._session.commit()
|