Select Git revision
org_round.py
-
Jiří Setnička authoredJiří Setnička authored
db.py 26.68 KiB
# SQLAlchemy definitions of all tables in the database
# Generated by sqlacodegen and then heavily edited.
import datetime
import decimal
from enum import Enum as PythonEnum, auto
import locale
import re
from sqlalchemy import \
Boolean, Column, DateTime, ForeignKey, Integer, String, Text, UniqueConstraint, \
text, func, \
create_engine, inspect, select
from sqlalchemy.engine import Engine
from sqlalchemy.orm import relationship, sessionmaker, Session, class_mapper, joinedload
from sqlalchemy.orm.attributes import get_history
from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.sql.expression import CTE
from sqlalchemy.sql.sqltypes import Numeric
from typing import Optional, List, Tuple
import mo
from mo.place_level import place_levels, PlaceLevel
from mo.util_format import timedelta, time_and_timedelta
# HACK: Work-around for https://github.com/dropbox/sqlalchemy-stubs/issues/114
from typing import TYPE_CHECKING, TypeVar, Type, Any
if TYPE_CHECKING:
from sqlalchemy.sql.type_api import TypeEngine
T = TypeVar('T')
class Enum(TypeEngine[T]):
def __init__(self, enum: Type[T], **kwargs: Any) -> None: ...
else:
from sqlalchemy import Enum
Base = declarative_base()
metadata = Base.metadata
class MOEnum(str, PythonEnum):
"""MOEnum je varianta PythonEnum, ve které se automaticky přidělované
hodnoty jmenují stejně jako klíče a funguje serializace do JSONu."""
def _generate_next_value_(name, start, count, last_values):
return name
@classmethod
def choices(enum):
out = []
for item in enum:
out.append((item.name, item.friendly_name()))
return out
@classmethod
def coerce(enum, name):
if isinstance(name, enum):
return name
try:
return enum[name]
except KeyError:
raise ValueError(name)
class PlaceType(MOEnum):
region = auto()
school = auto()
site = auto()
@classmethod
def choices(enum, level=None):
out = []
for item in enum:
(name, levels) = place_type_names_and_levels[item]
if level is None or level in levels:
if item == enum.region and level is not None:
name += " (" + place_levels[level].name + ")"
out.append((item.name, name))
return out
# list of allowed levels
place_type_names_and_levels = {
PlaceType.region: ('Region', [0, 1, 2, 3]),
PlaceType.school: ('Škola', [4]),
PlaceType.site: ('Soutěžní místo', [4]),
}
class Place(Base):
__tablename__ = 'places'
place_id = Column(Integer, primary_key=True, server_default=text("nextval('places_place_id_seq'::regclass)"))
level = Column(Integer, nullable=False)
parent = Column(Integer, ForeignKey('places.place_id'))
name = Column(String(255))
code = Column(String(255))
type = Column(Enum(PlaceType, name='place_type'), nullable=False)
nuts = Column(String(255), unique=True, server_default=text("NULL::character varying"))
note = Column(Text, nullable=False, server_default=text("''::text"))
children = relationship('Place')
school = relationship('School', uselist=False, back_populates='place')
def type_name(self):
if self.type == PlaceType.site:
return "soutěžní místo"
elif self.type == PlaceType.school:
return "škola"
elif self.level < len(place_levels):
return place_levels[self.level].name
else:
return "region"
def get_code(self):
return self.code or '#' + str(self.place_id)
def can_have_child(self):
return len(PlaceType.choices(level=self.level + 1)) > 0
def get_level(self) -> PlaceLevel:
return place_levels[self.level]
def name_locative(self):
name = self.name
if self.level == 1:
name = name.replace("ý kraj", "ém").replace("Kraj ", "")
return place_levels[self.level].in_name() + " " + name
def get_root_place():
return get_session().query(Place).filter_by(parent=None).one()
def get_place_by_code(code: str, fetch_school: bool = False) -> Optional[Place]:
code = code.strip()
if code == "":
return None
q = get_session().query(Place)
if fetch_school:
q = q.options(joinedload(Place.school))
m = re.fullmatch(r'#(\d+)', code)
if m:
return q.get(int(m[1]))
else:
return q.filter_by(code=code).one_or_none()
class School(Base):
__tablename__ = 'schools'
place_id = Column(Integer, ForeignKey('places.place_id', ondelete='CASCADE'), primary_key=True)
red_izo = Column(String(255), server_default=text("NULL::character varying"))
ico = Column(String(255), server_default=text("NULL::character varying"))
official_name = Column(String(255), server_default=text("NULL::character varying"))
address = Column(String(255), server_default=text("NULL::character varying"))
is_zs = Column(Boolean, nullable=False, server_default=text("false"))
is_ss = Column(Boolean, nullable=False, server_default=text("false"))
place = relationship('Place', uselist=False, back_populates='school')
class RoundState(MOEnum):
preparing = auto()
running = auto()
grading = auto()
closed = auto()
delegate = auto()
def friendly_name(self) -> str:
return round_state_names[self]
round_state_names = {
RoundState.preparing: 'připravuje se',
RoundState.running: 'běží',
RoundState.grading: 'opravuje se',
RoundState.closed: 'ukončeno',
RoundState.delegate: 'po oblastech',
}
class RoundScoreMode(MOEnum):
basic = auto()
mo = auto()
def friendly_name(self) -> str:
return round_score_mode_names[self]
round_score_mode_names = {
RoundScoreMode.basic: "Základní se sdílenými místy",
RoundScoreMode.mo: "Jednoznačné pořadí podle pravidel MO",
}
# V DB jako numeric(2,1), používá se tak snadněji, než enum
round_points_step_names = {
decimal.Decimal('1'): "Celé body",
decimal.Decimal('0.5'): "Půlbody",
decimal.Decimal('0.1'): "Desetinné body",
}
round_points_step_choices = round_points_step_names.items()
class Round(Base):
__tablename__ = 'rounds'
__table_args__ = (
UniqueConstraint('year', 'category', 'seq', 'part'),
)
round_id = Column(Integer, primary_key=True, server_default=text("nextval('rounds_round_id_seq'::regclass)"))
master_round_id = Column(Integer, ForeignKey('rounds.round_id'))
year = Column(Integer, nullable=False)
category = Column(String(2), nullable=False)
seq = Column(Integer, nullable=False)
part = Column(Integer, nullable=False)
level = Column(Integer, nullable=False)
name = Column(String(255), nullable=False)
state = Column(Enum(RoundState, name='round_state'), nullable=False, server_default=text("'preparing'::round_state"))
tasks_file = Column(String(255))
ct_tasks_start = Column(DateTime(True))
ct_submit_end = Column(DateTime(True))
pr_tasks_start = Column(DateTime(True))
pr_submit_end = Column(DateTime(True))
score_mode = Column(Enum(RoundScoreMode, name='score_mode'), nullable=False, server_default=text("'basic'::score_mode"))
score_winner_limit = Column(Numeric)
score_successful_limit = Column(Numeric)
points_step = Column(Numeric, nullable=False)
has_messages = Column(Boolean, nullable=False, server_default=text("false"))
master = relationship('Round', primaryjoin='Round.master_round_id == Round.round_id', remote_side='Round.round_id', post_update=True)
def round_code_short(self):
""" Pro samostatné kolo ekvivalentní s `round_code()`, pro skupinu kol společná část kódu. """
return f"{self.year}-{self.category}-{self.seq}"
def part_code(self):
return chr(ord('a') + self.part - 1) if self.part > 0 else ""
def round_code(self):
""" Kód kola včetně označení části, pokud je ve skupině kol. """
code = self.round_code_short()
part = self.part_code()
return f"{code}{part}"
def get_level(self) -> PlaceLevel:
return place_levels[self.level]
def has_tasks(self):
return self.tasks_file
def is_subround(self) -> bool:
return self.master_round_id != self.round_id
def get_group_rounds(self, add_self=False) -> List['Round']:
# podle PEP484 se má použít string při forward typové referenci
return get_session().query(Round).filter(
Round.master_round_id == self.master_round_id,
add_self or Round.round_id != self.round_id,
).all()
def ct_state(self) -> RoundState:
"""Vrátí stav z pohledu účastníků (vynucené 'preparing' před začátkem soutěže)."""
if not self.ct_tasks_start or self.ct_tasks_start > mo.now:
return RoundState.preparing
else:
return self.state
def points_step_name(self) -> str:
if self.points_step in round_points_step_names:
return round_points_step_names[self.points_step]
return str(self.points_step)
class User(Base):
__tablename__ = 'users'
user_id = Column(Integer, primary_key=True, server_default=text("nextval('users_user_id_seq'::regclass)"))
email = Column(String(255), nullable=False, unique=True)
first_name = Column(String(255), nullable=False)
last_name = Column(String(255), nullable=False)
is_org = Column(Boolean, nullable=False, server_default=text("false"))
is_admin = Column(Boolean, nullable=False, server_default=text("false"))
is_test = Column(Boolean, nullable=False, server_default=text("false"))
created_at = Column(DateTime(True), nullable=False, server_default=text("CURRENT_TIMESTAMP"))
last_login_at = Column(DateTime(True))
reset_at = Column(DateTime(True))
password_hash = Column(String(255), server_default=text("NULL::character varying"))
note = Column(Text, nullable=False, server_default=text("''::text"))
roles = relationship('UserRole', primaryjoin='UserRole.user_id == User.user_id')
participants = relationship('Participant', primaryjoin='Participant.user_id == User.user_id')
def full_name(self) -> str:
return self.first_name + ' ' + self.last_name
def sort_key(self) -> Tuple[str, str, int]:
return (locale.strxfrm(self.last_name), locale.strxfrm(self.first_name), self.user_id)
def name_sort_key(self) -> Tuple[str, str]:
return (locale.strxfrm(self.last_name), locale.strxfrm(self.first_name))
def is_inactive(self) -> bool:
return self.password_hash is None
class Contest(Base):
__tablename__ = 'contests'
__table_args__ = (
UniqueConstraint('round_id', 'place_id'),
)
contest_id = Column(Integer, primary_key=True, server_default=text("nextval('contests_contest_id_seq'::regclass)"))
master_contest_id = Column(Integer, ForeignKey('contests.contest_id'))
round_id = Column(Integer, ForeignKey('rounds.round_id'), nullable=False)
place_id = Column(Integer, ForeignKey('places.place_id'), nullable=False)
state = Column(Enum(RoundState, name='round_state'), nullable=False, server_default=text("'preparing'::round_state"))
master = relationship('Contest', primaryjoin='Contest.master_contest_id == Contest.contest_id', remote_side='Contest.contest_id', post_update=True)
place = relationship('Place')
round = relationship('Round')
def is_subcontest(self) -> bool:
return self.master_contest_id != self.contest_id
def get_group_contests(self, add_self=False) -> List['Contest']:
# podle PEP484 se má použít string při forward typové referenci
return get_session().query(Contest).filter(
Contest.master_contest_id == self.master_contest_id,
add_self or Contest.contest_id != self.contest_id,
).options(joinedload(Contest.round)).all()
def ct_state(self) -> RoundState:
"""Vrátí stav z pohledu účastníků (vynucené 'preparing' před začátkem soutěže)."""
state = self.round.ct_state()
if state != RoundState.delegate:
return state
else:
return self.state
def ct_long_state(self) -> str:
state = self.ct_state()
round = self.round
details = ""
if state == RoundState.preparing:
if round.ct_tasks_start and round.ct_tasks_start > mo.now:
details = f" – začne {time_and_timedelta(round.ct_tasks_start)}"
elif state == RoundState.running:
if round.ct_submit_end and round.ct_submit_end < mo.now:
details = f" – odevzdávání skončilo {time_and_timedelta(round.ct_submit_end)}"
elif round.ct_submit_end:
details = f" – odevzdávejte do {time_and_timedelta(round.ct_submit_end)}"
return state.friendly_name() + details
def ct_task_statement_available(self) -> bool:
"""Je zadaní dostupné pro účastníky?"""
return self.ct_state() != RoundState.preparing and self.round.tasks_file is not None
def ct_can_submit(self) -> bool:
"""Mohou účastníci odevzdávat?"""
return self.ct_state() == RoundState.running
class LogType(MOEnum):
general = auto()
user = auto()
place = auto()
round = auto()
contest = auto()
participant = auto()
task = auto()
user_role = auto()
class Log(Base):
__tablename__ = 'log'
log_entry_id = Column(Integer, primary_key=True, server_default=text("nextval('log_log_entry_id_seq'::regclass)"))
changed_by = Column(Integer, ForeignKey('users.user_id'))
changed_at = Column(DateTime(True), nullable=False, server_default=text("CURRENT_TIMESTAMP"))
type = Column(Enum(LogType, name='log_type'), nullable=False)
id = Column(Integer, nullable=False)
details = Column(JSONB, nullable=False)
user = relationship('User')
class Participant(Base):
__tablename__ = 'participants'
user_id = Column(Integer, ForeignKey('users.user_id'), primary_key=True, nullable=False)
year = Column(Integer, primary_key=True, nullable=False)
school = Column(Integer, ForeignKey('places.place_id'), nullable=False)
birth_year = Column(Integer, nullable=False)
grade = Column(String(20), nullable=False)
user = relationship('User')
school_place = relationship('Place', primaryjoin='Participant.school == Place.place_id')
class PartState(MOEnum):
registered = auto()
invited = auto()
refused = auto()
present = auto()
absent = auto()
disqualified = auto()
def friendly_name(self) -> str:
return part_state_names[self]
part_state_names = {
PartState.registered: 'přihlášený',
PartState.invited: 'pozvaný',
PartState.refused: 'odmítnutý',
PartState.present: 'přítomný',
PartState.absent: 'nepřítomný',
PartState.disqualified: 'diskvalifikovaný',
}
class Participation(Base):
__tablename__ = 'participations'
user_id = Column(Integer, ForeignKey('users.user_id'), primary_key=True, nullable=False)
contest_id = Column(Integer, ForeignKey('contests.contest_id'), primary_key=True, nullable=False)
place_id = Column(Integer, ForeignKey('places.place_id'), nullable=False)
state = Column(Enum(PartState, name='part_state'), nullable=False)
contest = relationship('Contest', primaryjoin='Participation.contest_id == Contest.contest_id')
place = relationship('Place', primaryjoin='Participation.place_id == Place.place_id')
user = relationship('User')
class Task(Base):
__tablename__ = 'tasks'
__table_args__ = (
UniqueConstraint('round_id', 'code'),
)
task_id = Column(Integer, primary_key=True, server_default=text("nextval('tasks_task_id_seq'::regclass)"))
round_id = Column(Integer, ForeignKey('rounds.round_id'), nullable=False)
code = Column(String(255), nullable=False)
name = Column(String(255), nullable=False)
max_points = Column(Numeric)
round = relationship('Round')
class RoleType(MOEnum):
garant = auto()
garant_kraj = auto()
garant_okres = auto()
garant_skola = auto()
dozor = auto()
opravovatel = auto()
def friendly_name(self) -> str:
return role_type_names[self]
role_type_names = {
RoleType.garant: 'celostátní garant',
RoleType.garant_kraj: 'krajský garant',
RoleType.garant_okres: 'okresní garant',
RoleType.garant_skola: 'školní garant',
RoleType.dozor: 'dozor',
RoleType.opravovatel: 'opravovatel',
}
class UserRole(Base):
__tablename__ = 'user_roles'
user_role_id = Column(Integer, primary_key=True, server_default=text("nextval('user_roles_user_role_id_seq'::regclass)"))
user_id = Column(Integer, ForeignKey('users.user_id'), nullable=False)
place_id = Column(Integer, ForeignKey('places.place_id'), nullable=False)
role = Column(Enum(RoleType, name='role_type'), nullable=False)
category = Column(String(2), server_default=text("NULL::character varying"))
year = Column(Integer)
seq = Column(Integer)
assigned_by = Column(Integer, ForeignKey('users.user_id'))
assigned_at = Column(DateTime(True), nullable=False, server_default=text("CURRENT_TIMESTAMP"))
user = relationship('User', primaryjoin='UserRole.user_id == User.user_id')
assigned_by_user = relationship('User', primaryjoin='UserRole.assigned_by == User.user_id')
place = relationship('Place')
def __str__(self):
parts = [self.role.friendly_name()]
if self.seq:
parts.append(f"{self.seq}. kola")
if self.year:
parts.append(f"{self.year}. ročníku")
if self.category:
parts.append(f"kategorie {self.category}")
if self.place.level > 0:
parts.append(self.place.name_locative())
return " ".join(parts)
class PaperType(MOEnum):
solution = auto()
feedback = auto()
class Paper(Base):
__tablename__ = 'papers'
paper_id = Column(Integer, primary_key=True, server_default=text("nextval('papers_paper_id_seq'::regclass)"))
for_task = Column(Integer, ForeignKey('tasks.task_id'), nullable=False)
for_user = Column(Integer, ForeignKey('users.user_id'), nullable=False)
type = Column(Enum(PaperType, name='paper_type'), nullable=False)
uploaded_by = Column(Integer, ForeignKey('users.user_id'), nullable=False)
uploaded_at = Column(DateTime(True), nullable=False, server_default=text("CURRENT_TIMESTAMP"))
pages = Column(Integer)
bytes = Column(Integer)
file_name = Column(String(255))
orig_file_name = Column(String(255))
note = Column(Text, nullable=False, server_default=text("''::text"))
task = relationship('Task')
for_user_obj = relationship('User', primaryjoin='Paper.for_user == User.user_id')
uploaded_by_obj = relationship('User', primaryjoin='Paper.uploaded_by == User.user_id')
def check_deadline(self, round: Round) -> Optional[str]:
# K round se dá dostat přes self.task.round, ale nejspíš to není nakešované.
if self.uploaded_by == self.for_user:
deadline = round.ct_submit_end
suffix = ' účastnickém termínu'
else:
deadline = round.pr_submit_end
suffix = ' dozorovém termínu'
if deadline is not None and self.uploaded_at > deadline:
return timedelta(self.uploaded_at, deadline, descriptive=True) + suffix
else:
return None
def is_broken(self) -> bool:
return self.file_name is None
def is_fixed(self) -> bool:
return self.orig_file_name is not None and self.file_name is not None
class PointsHistory(Base):
__tablename__ = 'points_history'
points_history_id = Column(Integer, primary_key=True, server_default=text("nextval('points_history_points_history_id_seq'::regclass)"))
task_id = Column(Integer, ForeignKey('tasks.task_id'), nullable=False)
participant_id = Column(Integer, ForeignKey('users.user_id'), nullable=False)
points = Column(Numeric)
points_by = Column(Integer, ForeignKey('users.user_id'), nullable=False)
points_at = Column(DateTime(True), nullable=False)
participant = relationship('User', primaryjoin='PointsHistory.participant_id == User.user_id')
user = relationship('User', primaryjoin='PointsHistory.points_by == User.user_id')
task = relationship('Task')
class Solution(Base):
__tablename__ = 'solutions'
task_id = Column(Integer, ForeignKey('tasks.task_id'), primary_key=True, nullable=False)
user_id = Column(Integer, ForeignKey('users.user_id'), primary_key=True, nullable=False)
final_submit = Column(Integer, ForeignKey('papers.paper_id'))
final_feedback = Column(Integer, ForeignKey('papers.paper_id'))
points = Column(Numeric)
note = Column(Text, nullable=False, server_default=text("''::text"))
org_note = Column(Text, nullable=False, server_default=text("''::text"))
final_submit_obj = relationship('Paper', primaryjoin='Solution.final_submit == Paper.paper_id')
final_feedback_obj = relationship('Paper', primaryjoin='Solution.final_feedback == Paper.paper_id')
task = relationship('Task')
user = relationship('User')
class JobType(MOEnum):
download_submits = auto()
upload_feedback = auto()
class JobState(MOEnum):
ready = auto()
running = auto()
done = auto()
failed = auto()
def friendly_name(self) -> str:
return job_state_names[self]
job_state_names = {
JobState.ready: 'čeká na spuštění',
JobState.running: 'zpracovává se',
JobState.done: 'dokončena',
JobState.failed: 'selhala',
}
class Job(Base):
__tablename__ = 'jobs'
job_id = Column(Integer, primary_key=True, server_default=text("nextval('jobs_job_id_seq'::regclass)"))
type = Column(Enum(JobType, name='job_type'), nullable=False)
state = Column(Enum(JobState, name='job_state'), nullable=False)
user_id = Column(Integer, ForeignKey('users.user_id'), nullable=False)
description = Column(Text, nullable=False, server_default=text("''::text"))
result = Column(Text, nullable=False, server_default=text("''::text"))
created_at = Column(DateTime(True), nullable=False, server_default=text("CURRENT_TIMESTAMP"))
finished_at = Column(DateTime(True))
expires_at = Column(DateTime(True))
in_json = Column(JSONB)
in_file = Column(String(255), server_default=text("NULL::character varying"))
out_json = Column(JSONB)
out_file = Column(String(255), server_default=text("NULL::character varying"))
user = relationship('User')
class Message(Base):
__tablename__ = 'messages'
message_id = Column(Integer, primary_key=True, server_default=text("nextval('messages_message_id_seq'::regclass)"))
round_id = Column(Integer, ForeignKey('rounds.round_id'), nullable=False)
created_at = Column(DateTime(True), nullable=False, server_default=text("CURRENT_TIMESTAMP"))
created_by = Column(Integer, ForeignKey('users.user_id'))
title = Column(Text, nullable=False)
markdown = Column(Text, nullable=False)
html = Column(Text, nullable=False)
created_by_user = relationship('User')
_engine: Optional[Engine] = None
_session: Optional[Session] = None
flask_db: Any = None
def get_session() -> Session:
global _session, _engine
if flask_db:
return flask_db.session
if _session is None:
if _engine is None:
import mo.config as config
_engine = create_engine(config.SQLALCHEMY_DATABASE_URI, echo=config.SQLALCHEMY_ECHO)
MOSession = sessionmaker(bind=_engine)
_session = MOSession()
return _session
def get_categories() -> List[str]:
return [cat for (cat,) in get_session().query(Round.category).distinct()]
def get_seqs() -> List[int]:
return [seq for (seq,) in get_session().query(Round.seq).distinct()]
def get_place_parents(place: Place) -> List[Place]:
"""Low-level funkce pro zjištění předků místa.
Obvykle voláme mo.rights.Gatekeeper.get_parents(), které kešuje."""
sess = get_session()
topq = (sess.query(Place)
.filter(Place.place_id == place.place_id)
.cte('parents', recursive=True))
botq = (sess.query(Place)
.join(topq, Place.place_id == topq.c.parent))
recq = topq.union(botq)
return sess.query(recq).all()
def place_descendant_cte(place: Place, max_level: Optional[int] = None) -> CTE:
"""Konstruuje CTE pro ID všech podřízených míst."""
sess = get_session()
topq = (sess.query(Place.place_id)
.filter(Place.place_id == place.place_id)
.cte('descendants', recursive=True))
botq = (sess.query(Place.place_id)
.join(topq, Place.parent == topq.c.place_id))
if max_level is not None:
botq = botq.filter(Place.level <= max_level)
return topq.union(botq)
def get_place_descendants(place: Place, min_level: Optional[int] = None, max_level: Optional[int] = None) -> List[Place]:
"""Zjištění všech podřízených míst v daném rozsahu úrovní."""
sess = get_session()
cte = place_descendant_cte(place, max_level)
q = sess.query(Place).filter(Place.place_id.in_(select([cte])))
if min_level is not None:
q = q.filter(Place.level >= min_level)
return q.all()
def get_object_changes(obj):
""" Given a model instance, returns dict of pending
changes waiting for database flush/commit.
e.g. {
'some_field': {
'before': *SOME-VALUE*,
'after': *SOME-VALUE*
},
...
}
Source: https://stackoverflow.com/questions/15952733/sqlalchemy-logging-of-changes-with-date-and-user
"""
inspection = inspect(obj)
changes = {}
for attr in class_mapper(obj.__class__).column_attrs:
if getattr(inspection.attrs, attr.key).history.has_changes():
if get_history(obj, attr.key)[2]:
before = get_history(obj, attr.key)[2].pop()
after = getattr(obj, attr.key)
if before != after:
if before or after:
changes[attr.key] = {'before': str(before), 'after': str(after)}
return changes
# FIXME: Asi přejmenovat, ať je jasné, že to je speciální funkce pro logování
def row2dict(row):
d = {}
for column in row.__table__.columns:
val = getattr(row, column.name)
if isinstance(val, datetime.datetime):
# datetime neumíme serializovat do JSONu, ale nevadí to, protože ho stejně nemá smysl logovat
pass
elif isinstance(val, decimal.Decimal):
d[column.name] = float(val)
else:
d[column.name] = getattr(row, column.name)
return d
# Rychlejší count než použití q.count() - viz https://gist.github.com/hest/8798884
# Varování: nepoužívat s podmínkami v lazyloadované části query, může vracet špatné
# výsledky!!!
def get_count(q):
return q.order_by(None).with_entities(func.count()).scalar()