Spaces:
Sleeping
Sleeping
File size: 3,439 Bytes
55d0d9e | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 | from sqlalchemy import select, func, or_, and_
from sqlalchemy.ext.asyncio import AsyncSession
from app.models.models import Definitions, CommitteeDesignation
import logging
class DefinitionService:
def __init__(self):
self.logger = logging.getLogger(__name__)
def _base_select(self):
return select(Definitions, CommitteeDesignation).outerjoin(
CommitteeDesignation,
Definitions.letter_tag == func.lower(CommitteeDesignation.letter_tag),
)
def _apply_filters(self, stmt, q: str | None, letter_tag: str | None):
conditions = []
if q and q.strip():
qq = f"%{q.strip()}%"
conditions.append(
or_(
Definitions.term.ilike(qq),
Definitions.definition.ilike(qq),
CommitteeDesignation.description.ilike(qq),
)
)
if letter_tag and letter_tag.strip():
lt = f"%{letter_tag.strip()}%"
conditions.append(
or_(
Definitions.letter_tag.ilike(lt),
CommitteeDesignation.letter_tag.ilike(lt),
)
)
if conditions:
return stmt.where(and_(*conditions))
return stmt
def row_to_dict(self, definition: Definitions, committee_designation):
return {
"definition": definition.definition or "",
"term": definition.term or "",
"letter_tag": definition.letter_tag or "",
"committee_designation": committee_designation.description
if committee_designation
else "No committee designation",
}
async def count_definitions(
self,
session: AsyncSession,
q: str | None = None,
letter_tag: str | None = None,
) -> int:
stmt = self._apply_filters(self._base_select(), q, letter_tag)
sub = stmt.subquery()
count_stmt = select(func.count()).select_from(sub)
result = await session.execute(count_stmt)
return int(result.scalar_one() or 0)
async def get_definitions_page(
self,
session: AsyncSession,
page: int,
page_size: int,
q: str | None = None,
letter_tag: str | None = None,
):
stmt = self._apply_filters(self._base_select(), q, letter_tag)
stmt = stmt.order_by(Definitions.term.asc()).offset(
(page - 1) * page_size
).limit(page_size)
result = await session.execute(stmt)
return result.all()
async def get_definitions_for_export(
self,
session: AsyncSession,
q: str | None = None,
letter_tag: str | None = None,
max_rows: int = 50_000,
):
stmt = self._apply_filters(self._base_select(), q, letter_tag)
stmt = stmt.order_by(Definitions.term.asc()).limit(max_rows)
result = await session.execute(stmt)
return result.all()
async def get_definitions(self, session: AsyncSession, limit: int):
"""Legacy: first N definitions ordered by term."""
return await self.get_definitions_page(session, 1, limit, None, None)
async def search_definition(self, session: AsyncSession, term: str):
return await self.get_definitions_page(
session, 1, 10_000, q=term, letter_tag=None
)
definition_service = DefinitionService()
|