93 lines
3.1 KiB
Python
93 lines
3.1 KiB
Python
import datetime
|
|
from typing import TypedDict
|
|
|
|
from folkugat_web.dal.sql import Connection, get_connection
|
|
from folkugat_web.model import sessions as model
|
|
from folkugat_web.model.sql import OrderCol, Range
|
|
from folkugat_web.typing import OptionalListOrValue as OLV
|
|
|
|
from . import conversion
|
|
|
|
|
|
class QueryData(TypedDict, total=False):
|
|
session_id: int | None
|
|
date_ub: datetime.date
|
|
date_lb: datetime.date
|
|
is_live: bool
|
|
limit: int
|
|
offset: int
|
|
|
|
|
|
def _filter_clause(session_id: int | None = None,
|
|
date_range: Range[datetime.date] | None = None,
|
|
is_live: bool | None = None) -> tuple[str, QueryData]:
|
|
filter_clauses: list[str] = []
|
|
filter_data: QueryData = {}
|
|
|
|
if session_id is not None:
|
|
filter_clauses.append(f"id = :session_id")
|
|
filter_data["session_id"] = session_id
|
|
|
|
if date_range:
|
|
if ub := date_range.upper_bound():
|
|
operator = "<=" if ub[1] else "<"
|
|
filter_clauses.append(f"date {operator} :date_ub")
|
|
filter_data["date_ub"] = ub[0]
|
|
if lb := date_range.lower_bound():
|
|
operator = ">=" if lb[1] else ">"
|
|
filter_clauses.append(f"date {operator} :date_lb")
|
|
filter_data["date_lb"] = lb[0]
|
|
|
|
if is_live is not None:
|
|
filter_clauses.append(f"is_live = :is_live")
|
|
filter_data["is_live"] = is_live
|
|
|
|
if filter_clauses:
|
|
filter_clause_str = " AND ".join(filter_clauses)
|
|
filter_clause = f"WHERE {filter_clause_str}"
|
|
return filter_clause, filter_data
|
|
else:
|
|
return "", {}
|
|
|
|
|
|
def _order_clause(order_by: OLV[OrderCol[model.SessionCols]]) -> str:
|
|
if not order_by:
|
|
return ""
|
|
if not isinstance(order_by, list):
|
|
order_by = [order_by]
|
|
|
|
order_clauses = [f"{ocol.column.value} {ocol.order.value}" for ocol in order_by]
|
|
order_clauses_str = " ".join(order_clauses)
|
|
return f"ORDER BY {order_clauses_str}"
|
|
|
|
|
|
def get_sessions(session_id: int | None = None,
|
|
date_range: Range[datetime.date] | None = None,
|
|
is_live: bool | None = None,
|
|
order_by: OLV[OrderCol[model.SessionCols]] = None,
|
|
limit: int | None = None, offset: int | None = None,
|
|
con: Connection | None = None) -> list[model.Session]:
|
|
clauses: list[str] = []
|
|
filter_clause, data = _filter_clause(session_id=session_id, date_range=date_range, is_live=is_live)
|
|
if filter_clause:
|
|
clauses.append(filter_clause)
|
|
if order_clause := _order_clause(order_by=order_by):
|
|
clauses.append(order_clause)
|
|
if limit is not None:
|
|
clauses.append("LIMIT :limit")
|
|
data["limit"] = limit
|
|
if offset is not None:
|
|
clauses.append("OFFSET :offset")
|
|
data["offset"] = offset
|
|
|
|
clauses_str = " ".join(clauses)
|
|
query = f"""
|
|
SELECT id, date, start_time, end_time, venue_name, venue_url, notes, cartell_url, is_live
|
|
FROM sessions
|
|
{clauses_str}
|
|
"""
|
|
with get_connection(con) as con:
|
|
cur = con.cursor()
|
|
_ = cur.execute(query, data)
|
|
return list(map(conversion.row_to_session, cur.fetchall()))
|