You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
58 lines
2.1 KiB
58 lines
2.1 KiB
from __future__ import annotations
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models.entities import Directory
|
|
|
|
|
|
class DirectoryRepository:
|
|
def __init__(self, session: AsyncSession) -> None:
|
|
self.session = session
|
|
|
|
async def get_by_id(self, directory_id: str) -> Directory | None:
|
|
return await self.session.get(Directory, directory_id)
|
|
|
|
async def get_root(self) -> Directory | None:
|
|
result = await self.session.execute(
|
|
select(Directory).where(Directory.parent_id.is_(None), Directory.deleted_at.is_(None))
|
|
)
|
|
return result.scalar_one_or_none()
|
|
|
|
async def list_children(self, parent_id: str) -> list[Directory]:
|
|
result = await self.session.execute(
|
|
select(Directory)
|
|
.where(Directory.parent_id == parent_id, Directory.deleted_at.is_(None))
|
|
.order_by(Directory.name.asc())
|
|
)
|
|
return list(result.scalars().all())
|
|
|
|
async def get_by_parent_and_name(self, parent_id: str | None, name: str) -> Directory | None:
|
|
result = await self.session.execute(
|
|
select(Directory).where(
|
|
Directory.parent_id == parent_id,
|
|
Directory.name == name,
|
|
Directory.deleted_at.is_(None),
|
|
)
|
|
)
|
|
return result.scalar_one_or_none()
|
|
|
|
async def list_descendants_by_path_prefix(self, path_prefix: str) -> list[Directory]:
|
|
like_pattern = f"{path_prefix}/%"
|
|
result = await self.session.execute(
|
|
select(Directory)
|
|
.where(Directory.path_key.like(like_pattern), Directory.deleted_at.is_(None))
|
|
.order_by(Directory.path_key.asc())
|
|
)
|
|
return list(result.scalars().all())
|
|
|
|
async def create(self, directory: Directory) -> Directory:
|
|
self.session.add(directory)
|
|
await self.session.flush()
|
|
await self.session.refresh(directory)
|
|
return directory
|
|
|
|
async def save(self, directory: Directory) -> Directory:
|
|
await self.session.flush()
|
|
await self.session.refresh(directory)
|
|
return directory
|
|
|