"""Database queries.""" import typing from sqlalchemy.orm.query import Query from sqlalchemy.engine.cursor import CursorResult from sqlalchemy import func from . import database, model def top_speakers() -> Query: """Find people who spoke at the most conferences.""" q: Query = ( database.session.query(model.Person, func.count()) .join(model.ConferencePerson) .filter(model.Person.id != 1046) # FOSDEM Staff .group_by(model.Person) .order_by(func.count().desc(), model.Person.name) ) return q def top_events() -> Query: """Most common titles of events.""" q: Query = ( database.session.query(model.Event.title, func.count()) .group_by(model.Event.title) .order_by(func.count().desc()) .having(func.count() > 3) ) return q def search_for_events(search_for: str) -> Query: """Search for events with by title.""" q: Query = model.Event.query.filter( model.Event.title.ilike(f"%{search_for}%") ).order_by(model.Event.title) return q def search_for_people(search_for: str) -> Query: """Search for people by name.""" q: Query = model.Person.query.filter( model.Person.name.ilike(f"%{search_for}%") ).order_by(model.Person.name) return q def speaker_counts() -> CursorResult: """Speaker/conference frequency distribution.""" sql = """ select num, count(*) from (select person_id, count(*) as num from conference_person group by person_id) a group by num order by num """ return typing.cast(CursorResult, database.session.execute(sql))