60 lines
1.6 KiB
Python
60 lines
1.6 KiB
Python
"""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))
|