41 lines
1.4 KiB
Python
41 lines
1.4 KiB
Python
"""Availability repository for database queries."""
|
|
|
|
from datetime import date
|
|
|
|
from sqlalchemy import and_, delete, select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from models import Availability
|
|
|
|
|
|
class AvailabilityRepository:
|
|
"""Repository for availability-related database queries."""
|
|
|
|
def __init__(self, db: AsyncSession):
|
|
self.db = db
|
|
|
|
async def get_by_date_range(
|
|
self, from_date: date, to_date: date
|
|
) -> list[Availability]:
|
|
"""Get availability slots for a date range."""
|
|
result = await self.db.execute(
|
|
select(Availability)
|
|
.where(and_(Availability.date >= from_date, Availability.date <= to_date))
|
|
.order_by(Availability.date, Availability.start_time)
|
|
)
|
|
return list(result.scalars().all())
|
|
|
|
async def get_by_date(self, target_date: date) -> list[Availability]:
|
|
"""Get availability slots for a specific date."""
|
|
result = await self.db.execute(
|
|
select(Availability)
|
|
.where(Availability.date == target_date)
|
|
.order_by(Availability.start_time)
|
|
)
|
|
return list(result.scalars().all())
|
|
|
|
async def delete_by_date(self, target_date: date) -> None:
|
|
"""Delete all availability for a specific date."""
|
|
await self.db.execute(
|
|
delete(Availability).where(Availability.date == target_date)
|
|
)
|