Skip to content
Snippets Groups Projects
Select Git revision
  • 06369dc4274bfceeaee967f1f8626f984c41e9fd
  • devel default
  • master
  • fo
  • jirka/typing
  • fo-base
  • mj/submit-images
  • jk/issue-96
  • jk/issue-196
  • honza/add-contestant
  • honza/mr7
  • honza/mrf
  • honza/mrd
  • honza/mra
  • honza/mr6
  • honza/submit-images
  • honza/kolo-vs-soutez
  • jh-stress-test-wip
  • shorten-schools
19 results

org_round.py

Blame
  • 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()