didier/database/crud/ufora_courses.py

39 lines
1.4 KiB
Python
Raw Normal View History

from typing import Optional
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
2022-08-29 20:24:42 +02:00
from database.schemas import UforaCourse, UforaCourseAlias
2022-09-17 23:20:46 +02:00
__all__ = ["get_all_courses", "get_course_by_code", "get_course_by_name"]
async def get_all_courses(session: AsyncSession) -> list[UforaCourse]:
"""Get a list of all courses in the database"""
statement = select(UforaCourse)
2022-07-14 22:03:56 +02:00
return list((await session.execute(statement)).scalars().all())
2022-09-17 23:20:46 +02:00
async def get_course_by_code(session: AsyncSession, code: str) -> Optional[UforaCourse]:
"""Try to find a course by its code"""
statement = select(UforaCourse).where(UforaCourse.code == code)
return (await session.execute(statement)).scalar_one_or_none()
async def get_course_by_name(session: AsyncSession, query: str) -> Optional[UforaCourse]:
"""Try to find a course by its name
This checks for regular name first, and then aliases
"""
# Search case-insensitively
query = query.lower()
2023-07-08 01:23:47 +02:00
course_statement = select(UforaCourse).where(UforaCourse.name.ilike(f"%{query}%"))
course_result = (await session.execute(course_statement)).scalars().first()
if course_result:
return course_result
alias_statement = select(UforaCourseAlias).where(UforaCourseAlias.alias.ilike(f"%{query}%"))
alias_result = (await session.execute(alias_statement)).scalars().first()
return alias_result.course if alias_result else None