Skip to content

Repositories

The repository layer lives in website/repositories/ and handles data access only — queries, CRUD, and database interactions.

Repositories:

  • Never contain business logic or validation
  • Never access Flask context
  • Never commit transactions (services own transaction boundaries)
  • Inherit from a shared BaseRepository for common CRUD operations

Overview

Repository Model Description
BaseRepository — Generic CRUD operations inherited by all repositories
ChannelRepository Channel Discord category management and size tracking
GameRepository Game Game queries with filtering, search, pagination, and eager loading
GameEventRepository GameEvent Game audit log entry creation
GameSessionRepository GameSession Session date range queries and conflict detection
SpecialEventRepository SpecialEvent Themed event retrieval with active/inactive filtering
SystemRepository System RPG system lookups
TrophyRepository Trophy, UserTrophy Achievement data and leaderboard aggregations
UserRepository User User entity retrieval
VttRepository Vtt Virtual tabletop platform lookups

API Reference

Repository layer for data access operations.

BaseRepository

Bases: Generic[T]

Base repository providing common CRUD operations.

Repositories never commit — they only add, flush, and delete. The service layer owns the transaction boundary.

Source code in website/repositories/base.py 
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
class BaseRepository(Generic[T]):
    """Base repository providing common CRUD operations.

    Repositories never commit — they only add, flush, and delete.
    The service layer owns the transaction boundary.
    """

    model_class: type[T]

    def __init__(self, session=None):
        self.session = session or db.session

    def get_by_id(self, id: int) -> T | None:
        """Retrieve an entity by its primary key.

        Args:
            id: Primary key value.

        Returns:
            Entity instance, or None if not found.
        """
        return self.session.get(self.model_class, id)

    def get_by_id_or_404(self, id: int) -> T:
        """Retrieve an entity by ID or abort with 404.

        Args:
            id: Primary key value.

        Returns:
            Entity instance.

        Raises:
            NotFound: If entity does not exist.
        """
        return db.get_or_404(self.model_class, id)

    def get_all(self) -> list[T]:
        """Retrieve all entities of this type.

        Returns:
            List of all entity instances.
        """
        return self.session.query(self.model_class).all()

    def add(self, entity: T) -> T:
        """Add an entity to the session and flush.

        Args:
            entity: Entity instance to persist.

        Returns:
            The persisted entity.
        """
        self.session.add(entity)
        self.session.flush()
        return entity

    def delete(self, entity: T) -> None:
        """Delete an entity from the session and flush.

        Args:
            entity: Entity instance to remove.
        """
        self.session.delete(entity)
        self.session.flush()

    def count(self) -> int:
        """Return the total count of entities.

        Returns:
            Integer count.
        """
        return self.session.query(self.model_class).count()

get_by_id(id)

Retrieve an entity by its primary key.

Parameters:

Name Type Description Default
id int

Primary key value.

 required

Returns:

Type Description
T | None

Entity instance, or None if not found.
Source code in website/repositories/base.py 
22
23
24
25
26
27
28
29
30
31
def get_by_id(self, id: int) -> T | None:
    """Retrieve an entity by its primary key.

    Args:
        id: Primary key value.

    Returns:
        Entity instance, or None if not found.
    """
    return self.session.get(self.model_class, id)

get_by_id_or_404(id)

Retrieve an entity by ID or abort with 404.

Parameters:

Name Type Description Default
id int

Primary key value.

 required

Returns:

Type Description
T

Entity instance.

Raises:

Type Description
NotFound

If entity does not exist.
Source code in website/repositories/base.py 
33
34
35
36
37
38
39
40
41
42
43
44
45
def get_by_id_or_404(self, id: int) -> T:
    """Retrieve an entity by ID or abort with 404.

    Args:
        id: Primary key value.

    Returns:
        Entity instance.

    Raises:
        NotFound: If entity does not exist.
    """
    return db.get_or_404(self.model_class, id)

get_all()

Retrieve all entities of this type.

Returns:

Type Description
list[T]

List of all entity instances.
Source code in website/repositories/base.py 
47
48
49
50
51
52
53
def get_all(self) -> list[T]:
    """Retrieve all entities of this type.

    Returns:
        List of all entity instances.
    """
    return self.session.query(self.model_class).all()

add(entity)

Add an entity to the session and flush.

Parameters:

Name Type Description Default
entity T

Entity instance to persist.

 required

Returns:

Type Description
T

The persisted entity.
Source code in website/repositories/base.py 
55
56
57
58
59
60
61
62
63
64
65
66
def add(self, entity: T) -> T:
    """Add an entity to the session and flush.

    Args:
        entity: Entity instance to persist.

    Returns:
        The persisted entity.
    """
    self.session.add(entity)
    self.session.flush()
    return entity

delete(entity)

Delete an entity from the session and flush.

Parameters:

Name Type Description Default
entity T

Entity instance to remove.

 required
Source code in website/repositories/base.py 
68
69
70
71
72
73
74
75
def delete(self, entity: T) -> None:
    """Delete an entity from the session and flush.

    Args:
        entity: Entity instance to remove.
    """
    self.session.delete(entity)
    self.session.flush()

count()

Return the total count of entities.

Returns:

Type Description
int

Integer count.
Source code in website/repositories/base.py 
77
78
79
80
81
82
83
def count(self) -> int:
    """Return the total count of entities.

    Returns:
        Integer count.
    """
    return self.session.query(self.model_class).count()

ChannelRepository

Bases: BaseRepository[Channel]

Repository for Channel (Discord category) entities.

Source code in website/repositories/channel.py 
 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
class ChannelRepository(BaseRepository[Channel]):
    """Repository for Channel (Discord category) entities."""

    model_class = Channel

    def get_smallest_by_type(self, type: str) -> Channel | None:
        """Find the category with the fewest channels for a game type.

        Args:
            type: Game type (oneshot or campaign).

        Returns:
            Channel with smallest size, or None if no match.
        """
        return self.session.query(Channel).filter_by(type=type).order_by(Channel.size).first()

    def increment_size(self, channel: Channel) -> None:
        """Increment the channel count of a category.

        Args:
            channel: Channel entity to update.
        """
        channel.size += 1
        self.session.flush()

    def decrement_size(self, channel: Channel) -> None:
        """Decrement the channel count of a category (minimum 0).

        Args:
            channel: Channel entity to update.
        """
        channel.size = max(0, channel.size - 1)
        self.session.flush()

get_smallest_by_type(type)

Find the category with the fewest channels for a game type.

Parameters:

Name Type Description Default
type str

Game type (oneshot or campaign).

 required

Returns:

Type Description
Channel | None

Channel with smallest size, or None if no match.
Source code in website/repositories/channel.py 
12
13
14
15
16
17
18
19
20
21
def get_smallest_by_type(self, type: str) -> Channel | None:
    """Find the category with the fewest channels for a game type.

    Args:
        type: Game type (oneshot or campaign).

    Returns:
        Channel with smallest size, or None if no match.
    """
    return self.session.query(Channel).filter_by(type=type).order_by(Channel.size).first()

increment_size(channel)

Increment the channel count of a category.

Parameters:

Name Type Description Default
channel Channel

Channel entity to update.

 required
Source code in website/repositories/channel.py 
23
24
25
26
27
28
29
30
def increment_size(self, channel: Channel) -> None:
    """Increment the channel count of a category.

    Args:
        channel: Channel entity to update.
    """
    channel.size += 1
    self.session.flush()

decrement_size(channel)

Decrement the channel count of a category (minimum 0).

Parameters:

Name Type Description Default
channel Channel

Channel entity to update.

 required
Source code in website/repositories/channel.py 
32
33
34
35
36
37
38
39
def decrement_size(self, channel: Channel) -> None:
    """Decrement the channel count of a category (minimum 0).

    Args:
        channel: Channel entity to update.
    """
    channel.size = max(0, channel.size - 1)
    self.session.flush()

GameRepository

Bases: BaseRepository[Game]

Repository for Game entity.

Provides data access methods for games including queries, filtering, and search. All methods return models or None - exceptions for 404s should be raised by services.

Source code in website/repositories/game.py 
 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
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
class GameRepository(BaseRepository[Game]):
    """Repository for Game entity.

    Provides data access methods for games including queries, filtering, and search.
    All methods return models or None - exceptions for 404s should be raised by services.
    """

    model_class = Game

    def get_by_slug(self, slug: str) -> Optional[Game]:
        """Get game by slug.

        Args:
            slug: URL-safe game identifier.

        Returns:
            Game instance or None if not found.
        """
        return self.session.query(Game).filter_by(slug=slug).first()

    def get_by_slug_or_404(self, slug: str) -> Game:
        """Get game by slug or raise 404.

        Args:
            slug: URL-safe game identifier.

        Returns:
            Game instance.

        Raises:
            NotFound: If game with slug doesn't exist.
        """
        return self.session.query(Game).filter_by(slug=slug).first_or_404()

    def get_all_slugs(self) -> set[str]:
        """Get all existing game slugs.

        Returns:
            Set of all game slugs.
        """
        return {g.slug for g in self.session.query(Game.slug).all()}

    def find_by_gm(self, gm_id: str) -> list[Game]:
        """Find all games by GM.

        Args:
            gm_id: GM user ID.

        Returns:
            List of games GMed by this user.
        """
        return self.session.query(Game).filter_by(gm_id=gm_id).all()

    def find_by_player(self, player_id: str) -> list[Game]:
        """Find all games where user is a player.

        Args:
            player_id: Player user ID.

        Returns:
            List of games where user is registered as player.
        """
        return self.session.query(Game).join(Game.players).filter(User.id == player_id).all()

    def find_by_special_event(self, event_id: int) -> list[Game]:
        """Find all games for a special event.

        Args:
            event_id: Special event ID.

        Returns:
            List of games linked to this special event.
        """
        return self.session.query(Game).filter_by(special_event_id=event_id).all()

    def query_by_special_event(self, event_id: int):
        """Return a query object for games linked to a special event.

        Unlike ``find_by_special_event`` (which returns a list), this returns
        an unevaluated query suitable for further filtering and pagination.

        Args:
            event_id: Special event ID.

        Returns:
            SQLAlchemy query filtered by special_event_id.
        """
        return self.session.query(Game).filter(Game.special_event_id == event_id)

    def get_for_update(self, game_id: int) -> Optional[Game]:
        """Get game with pessimistic lock for updates.

        Uses SELECT FOR UPDATE to prevent race conditions during registration.

        Args:
            game_id: Game ID to lock.

        Returns:
            Game instance with exclusive lock, or None if not found.
        """
        return self.session.query(Game).filter_by(id=game_id).with_for_update().first()

    def get_with_relations(self, game_id: int) -> Optional[Game]:
        """Get game with all relationships eagerly loaded.

        Prevents N+1 queries by loading gm, system, vtt, players, and sessions.

        Args:
            game_id: Game ID.

        Returns:
            Game instance with relationships loaded, or None if not found.
        """
        return (
            self.session.query(Game)
            .options(
                joinedload(Game.gm),
                joinedload(Game.system),
                joinedload(Game.vtt),
                subqueryload(Game.players),
                subqueryload(Game.sessions),
            )
            .filter_by(id=game_id)
            .first()
        )

    def delete_by_id(self, game_id: int) -> None:
        """Delete game by ID.

        Args:
            game_id: Game ID to delete.
        """
        self.session.query(Game).filter_by(id=game_id).delete()
        self.session.flush()

    def search(
        self,
        filters: dict,
        page: int = 1,
        per_page: int = 20,
        user_payload: Optional[dict] = None,
    ) -> tuple[list[Game], int]:
        """Search games with complex filters and pagination.

        Replaces the complex get_filtered_games logic from helpers.

        Args:
            filters: Dict containing:
                - status: List of statuses (open, closed, archived, draft)
                - game_type: List of types (oneshot, campaign)
                - restriction: List of restrictions (all, 16+, 18+)
                - name: Optional name search string
                - system_id: Optional system ID
                - vtt_id: Optional VTT ID
                - gm_id: Optional GM ID (for "my games" views)
                - player_id: Optional player ID (for "my games as player" views)
                - special_event_id: Optional special event ID
            page: Page number (1-indexed).
            per_page: Items per page.
            user_payload: Optional user auth payload for permission filtering.

        Returns:
            Tuple of (games list, total count).
        """
        now = datetime.now(timezone.utc)
        query = self.session.query(Game)

        # Status filter with permission check
        status = filters.get("status", ["open"])
        if status:
            status_filters = []
            for s in status:
                if s != "draft":
                    status_filters.append(Game.status == s)
                elif user_payload and user_payload.get("is_admin"):
                    status_filters.append(Game.status == "draft")
                elif user_payload:
                    status_filters.append(
                        and_(
                            Game.status == "draft",
                            Game.gm_id == user_payload.get("user_id"),
                        )
                    )
            if status_filters:
                query = query.filter(or_(*status_filters))

        # Type filter
        game_type = filters.get("game_type", ["oneshot", "campaign"])
        if game_type:
            query = query.filter(Game.type.in_(game_type))

        # Restriction filter
        restriction = filters.get("restriction", ["all", "16+", "18+"])
        if restriction:
            query = query.filter(Game.restriction.in_(restriction))

        # Name search
        if filters.get("name"):
            query = query.filter(Game.name.ilike(f"%{filters['name']}%"))

        # System filter
        if filters.get("system_id"):
            query = query.filter(Game.system_id == filters["system_id"])

        # VTT filter
        if filters.get("vtt_id"):
            query = query.filter(Game.vtt_id == filters["vtt_id"])

        # GM filter
        if filters.get("gm_id"):
            query = query.filter(Game.gm_id == filters["gm_id"])

        # Player filter
        if filters.get("player_id"):
            query = query.join(Game.players).filter(User.id == filters["player_id"])

        # Special event filter
        if filters.get("special_event_id"):
            query = query.filter(Game.special_event_id == filters["special_event_id"])

        # Sorting - same logic as original helpers.py
        status_order = case(
            (Game.status == "draft", 0),
            (Game.status == "open", 1),
            (Game.status == "closed", 2),
            (Game.status == "archived", 3),
        )
        is_future = case((Game.date >= now, 0), else_=1)
        time_distance = func.abs(func.extract("epoch", Game.date - now))

        query = query.order_by(status_order, is_future, time_distance)

        # Get total count before pagination
        total = query.count()

        # Pagination
        offset = (page - 1) * per_page
        games = query.limit(per_page).offset(offset).all()

        return games, total

get_by_slug(slug)

Get game by slug.

Parameters:

Name Type Description Default
slug str

URL-safe game identifier.

 required

Returns:

Type Description
Optional[Game]

Game instance or None if not found.
Source code in website/repositories/game.py 
22
23
24
25
26
27
28
29
30
31
def get_by_slug(self, slug: str) -> Optional[Game]:
    """Get game by slug.

    Args:
        slug: URL-safe game identifier.

    Returns:
        Game instance or None if not found.
    """
    return self.session.query(Game).filter_by(slug=slug).first()

get_by_slug_or_404(slug)

Get game by slug or raise 404.

Parameters:

Name Type Description Default
slug str

URL-safe game identifier.

 required

Returns:

Type Description
Game

Game instance.

Raises:

Type Description
NotFound

If game with slug doesn't exist.
Source code in website/repositories/game.py 
33
34
35
36
37
38
39
40
41
42
43
44
45
def get_by_slug_or_404(self, slug: str) -> Game:
    """Get game by slug or raise 404.

    Args:
        slug: URL-safe game identifier.

    Returns:
        Game instance.

    Raises:
        NotFound: If game with slug doesn't exist.
    """
    return self.session.query(Game).filter_by(slug=slug).first_or_404()

get_all_slugs()

Get all existing game slugs.

Returns:

Type Description
set[str]

Set of all game slugs.
Source code in website/repositories/game.py 
47
48
49
50
51
52
53
def get_all_slugs(self) -> set[str]:
    """Get all existing game slugs.

    Returns:
        Set of all game slugs.
    """
    return {g.slug for g in self.session.query(Game.slug).all()}

find_by_gm(gm_id)

Find all games by GM.

Parameters:

Name Type Description Default
gm_id str

GM user ID.

 required

Returns:

Type Description
list[Game]

List of games GMed by this user.
Source code in website/repositories/game.py 
55
56
57
58
59
60
61
62
63
64
def find_by_gm(self, gm_id: str) -> list[Game]:
    """Find all games by GM.

    Args:
        gm_id: GM user ID.

    Returns:
        List of games GMed by this user.
    """
    return self.session.query(Game).filter_by(gm_id=gm_id).all()

find_by_player(player_id)

Find all games where user is a player.

Parameters:

Name Type Description Default
player_id str

Player user ID.

 required

Returns:

Type Description
list[Game]

List of games where user is registered as player.
Source code in website/repositories/game.py 
66
67
68
69
70
71
72
73
74
75
def find_by_player(self, player_id: str) -> list[Game]:
    """Find all games where user is a player.

    Args:
        player_id: Player user ID.

    Returns:
        List of games where user is registered as player.
    """
    return self.session.query(Game).join(Game.players).filter(User.id == player_id).all()

find_by_special_event(event_id)

Find all games for a special event.

Parameters:

Name Type Description Default
event_id int

Special event ID.

 required

Returns:

Type Description
list[Game]

List of games linked to this special event.
Source code in website/repositories/game.py 
77
78
79
80
81
82
83
84
85
86
def find_by_special_event(self, event_id: int) -> list[Game]:
    """Find all games for a special event.

    Args:
        event_id: Special event ID.

    Returns:
        List of games linked to this special event.
    """
    return self.session.query(Game).filter_by(special_event_id=event_id).all()

query_by_special_event(event_id)

Return a query object for games linked to a special event.

Unlike find_by_special_event (which returns a list), this returns an unevaluated query suitable for further filtering and pagination.

Parameters:

Name Type Description Default
event_id int

Special event ID.

 required

Returns:

Type Description

SQLAlchemy query filtered by special_event_id.
Source code in website/repositories/game.py 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
def query_by_special_event(self, event_id: int):
    """Return a query object for games linked to a special event.

    Unlike ``find_by_special_event`` (which returns a list), this returns
    an unevaluated query suitable for further filtering and pagination.

    Args:
        event_id: Special event ID.

    Returns:
        SQLAlchemy query filtered by special_event_id.
    """
    return self.session.query(Game).filter(Game.special_event_id == event_id)

get_for_update(game_id)

Get game with pessimistic lock for updates.

Uses SELECT FOR UPDATE to prevent race conditions during registration.

Parameters:

Name Type Description Default
game_id int

Game ID to lock.

 required

Returns:

Type Description
Optional[Game]

Game instance with exclusive lock, or None if not found.
Source code in website/repositories/game.py 
102
103
104
105
106
107
108
109
110
111
112
113
def get_for_update(self, game_id: int) -> Optional[Game]:
    """Get game with pessimistic lock for updates.

    Uses SELECT FOR UPDATE to prevent race conditions during registration.

    Args:
        game_id: Game ID to lock.

    Returns:
        Game instance with exclusive lock, or None if not found.
    """
    return self.session.query(Game).filter_by(id=game_id).with_for_update().first()

get_with_relations(game_id)

Get game with all relationships eagerly loaded.

Prevents N+1 queries by loading gm, system, vtt, players, and sessions.

Parameters:

Name Type Description Default
game_id int

Game ID.

 required

Returns:

Type Description
Optional[Game]

Game instance with relationships loaded, or None if not found.
Source code in website/repositories/game.py 
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
def get_with_relations(self, game_id: int) -> Optional[Game]:
    """Get game with all relationships eagerly loaded.

    Prevents N+1 queries by loading gm, system, vtt, players, and sessions.

    Args:
        game_id: Game ID.

    Returns:
        Game instance with relationships loaded, or None if not found.
    """
    return (
        self.session.query(Game)
        .options(
            joinedload(Game.gm),
            joinedload(Game.system),
            joinedload(Game.vtt),
            subqueryload(Game.players),
            subqueryload(Game.sessions),
        )
        .filter_by(id=game_id)
        .first()
    )

delete_by_id(game_id)

Delete game by ID.

Parameters:

Name Type Description Default
game_id int

Game ID to delete.

 required
Source code in website/repositories/game.py 
139
140
141
142
143
144
145
146
def delete_by_id(self, game_id: int) -> None:
    """Delete game by ID.

    Args:
        game_id: Game ID to delete.
    """
    self.session.query(Game).filter_by(id=game_id).delete()
    self.session.flush()

search(filters, page=1, per_page=20, user_payload=None)

Search games with complex filters and pagination.

Replaces the complex get_filtered_games logic from helpers.

Parameters:

Name Type Description Default
filters dict

Dict containing: - status: List of statuses (open, closed, archived, draft) - game_type: List of types (oneshot, campaign) - restriction: List of restrictions (all, 16+, 18+) - name: Optional name search string - system_id: Optional system ID - vtt_id: Optional VTT ID - gm_id: Optional GM ID (for "my games" views) - player_id: Optional player ID (for "my games as player" views) - special_event_id: Optional special event ID

 required
page int

Page number (1-indexed).

 1
per_page int

Items per page.

 20
user_payload Optional[dict]

Optional user auth payload for permission filtering.

 None

Returns:

Type Description
tuple[list[Game], int]

Tuple of (games list, total count).
Source code in website/repositories/game.py 
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
def search(
    self,
    filters: dict,
    page: int = 1,
    per_page: int = 20,
    user_payload: Optional[dict] = None,
) -> tuple[list[Game], int]:
    """Search games with complex filters and pagination.

    Replaces the complex get_filtered_games logic from helpers.

    Args:
        filters: Dict containing:
            - status: List of statuses (open, closed, archived, draft)
            - game_type: List of types (oneshot, campaign)
            - restriction: List of restrictions (all, 16+, 18+)
            - name: Optional name search string
            - system_id: Optional system ID
            - vtt_id: Optional VTT ID
            - gm_id: Optional GM ID (for "my games" views)
            - player_id: Optional player ID (for "my games as player" views)
            - special_event_id: Optional special event ID
        page: Page number (1-indexed).
        per_page: Items per page.
        user_payload: Optional user auth payload for permission filtering.

    Returns:
        Tuple of (games list, total count).
    """
    now = datetime.now(timezone.utc)
    query = self.session.query(Game)

    # Status filter with permission check
    status = filters.get("status", ["open"])
    if status:
        status_filters = []
        for s in status:
            if s != "draft":
                status_filters.append(Game.status == s)
            elif user_payload and user_payload.get("is_admin"):
                status_filters.append(Game.status == "draft")
            elif user_payload:
                status_filters.append(
                    and_(
                        Game.status == "draft",
                        Game.gm_id == user_payload.get("user_id"),
                    )
                )
        if status_filters:
            query = query.filter(or_(*status_filters))

    # Type filter
    game_type = filters.get("game_type", ["oneshot", "campaign"])
    if game_type:
        query = query.filter(Game.type.in_(game_type))

    # Restriction filter
    restriction = filters.get("restriction", ["all", "16+", "18+"])
    if restriction:
        query = query.filter(Game.restriction.in_(restriction))

    # Name search
    if filters.get("name"):
        query = query.filter(Game.name.ilike(f"%{filters['name']}%"))

    # System filter
    if filters.get("system_id"):
        query = query.filter(Game.system_id == filters["system_id"])

    # VTT filter
    if filters.get("vtt_id"):
        query = query.filter(Game.vtt_id == filters["vtt_id"])

    # GM filter
    if filters.get("gm_id"):
        query = query.filter(Game.gm_id == filters["gm_id"])

    # Player filter
    if filters.get("player_id"):
        query = query.join(Game.players).filter(User.id == filters["player_id"])

    # Special event filter
    if filters.get("special_event_id"):
        query = query.filter(Game.special_event_id == filters["special_event_id"])

    # Sorting - same logic as original helpers.py
    status_order = case(
        (Game.status == "draft", 0),
        (Game.status == "open", 1),
        (Game.status == "closed", 2),
        (Game.status == "archived", 3),
    )
    is_future = case((Game.date >= now, 0), else_=1)
    time_distance = func.abs(func.extract("epoch", Game.date - now))

    query = query.order_by(status_order, is_future, time_distance)

    # Get total count before pagination
    total = query.count()

    # Pagination
    offset = (page - 1) * per_page
    games = query.limit(per_page).offset(offset).all()

    return games, total

GameEventRepository

Bases: BaseRepository[GameEvent]

Repository for GameEvent entities.

Source code in website/repositories/game_event.py 
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
class GameEventRepository(BaseRepository[GameEvent]):
    """Repository for GameEvent entities."""

    model_class = GameEvent

    def log(
        self, action: str, game_id: int, description: str | None = None, user_id: str | None = None
    ) -> GameEvent:
        """Create and persist a new game event.

        Args:
            action: Event action type.
            game_id: ID of the related game.
            description: Optional event description.
            user_id: Optional ID of the user that performed the action.

        Returns:
            Created GameEvent instance.
        """
        event = GameEvent(action=action, game_id=game_id, description=description, user_id=user_id)
        return self.add(event)

log(action, game_id, description=None, user_id=None)

Create and persist a new game event.

Parameters:

Name Type Description Default
action str

Event action type.

 required
game_id int

ID of the related game.

 required
description str | None

Optional event description.

 None
user_id str | None

Optional ID of the user that performed the action.

 None

Returns:

Type Description
GameEvent

Created GameEvent instance.
Source code in website/repositories/game_event.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
def log(
    self, action: str, game_id: int, description: str | None = None, user_id: str | None = None
) -> GameEvent:
    """Create and persist a new game event.

    Args:
        action: Event action type.
        game_id: ID of the related game.
        description: Optional event description.
        user_id: Optional ID of the user that performed the action.

    Returns:
        Created GameEvent instance.
    """
    event = GameEvent(action=action, game_id=game_id, description=description, user_id=user_id)
    return self.add(event)

GameSessionRepository

Bases: BaseRepository[GameSession]

Repository for GameSession entities.

Source code in website/repositories/game_session.py 
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
class GameSessionRepository(BaseRepository[GameSession]):
    """Repository for GameSession entities."""

    model_class = GameSession

    def find_in_range(self, start: datetime, end: datetime) -> list[GameSession]:
        """Find all sessions within a date range.

        Args:
            start: Range start datetime (inclusive).
            end: Range end datetime (inclusive).

        Returns:
            List of GameSession instances within the range.
        """
        return (
            self.session.query(GameSession)
            .filter(GameSession.start >= start, GameSession.end <= end)
            .all()
        )

find_in_range(start, end)

Find all sessions within a date range.

Parameters:

Name Type Description Default
start datetime

Range start datetime (inclusive).

 required
end datetime

Range end datetime (inclusive).

 required

Returns:

Type Description
list[GameSession]

List of GameSession instances within the range.
Source code in website/repositories/game_session.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
def find_in_range(self, start: datetime, end: datetime) -> list[GameSession]:
    """Find all sessions within a date range.

    Args:
        start: Range start datetime (inclusive).
        end: Range end datetime (inclusive).

    Returns:
        List of GameSession instances within the range.
    """
    return (
        self.session.query(GameSession)
        .filter(GameSession.start >= start, GameSession.end <= end)
        .all()
    )

SpecialEventRepository

Bases: BaseRepository[SpecialEvent]

Repository for SpecialEvent data access.

Handles querying special events with optional filtering by active status.

Source code in website/repositories/special_event.py 
 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
class SpecialEventRepository(BaseRepository[SpecialEvent]):
    """Repository for SpecialEvent data access.

    Handles querying special events with optional filtering by active status.
    """

    model_class = SpecialEvent

    def get_all(self, active_only: bool = False) -> list[SpecialEvent]:
        """Get all special events, optionally filtered by active status.

        Args:
            active_only: If True, only return active events. Defaults to False.

        Returns:
            List of SpecialEvent instances ordered by name.
        """
        query = self.session.query(SpecialEvent).order_by(SpecialEvent.name)
        if active_only:
            query = query.filter_by(active=True)
        return query.all()

    def get_active(self) -> list[SpecialEvent]:
        """Get all active special events.

        Convenience method for getting active events only.

        Returns:
            List of active SpecialEvent instances ordered by name.
        """
        return self.get_all(active_only=True)

    def get_by_name(self, name: str) -> SpecialEvent | None:
        """Get special event by name.

        Args:
            name: Name of the special event.

        Returns:
            SpecialEvent instance if found, None otherwise.
        """
        return self.session.query(SpecialEvent).filter_by(name=name).first()

get_all(active_only=False)

Get all special events, optionally filtered by active status.

Parameters:

Name Type Description Default
active_only bool

If True, only return active events. Defaults to False.

 False

Returns:

Type Description
list[SpecialEvent]

List of SpecialEvent instances ordered by name.
Source code in website/repositories/special_event.py 
15
16
17
18
19
20
21
22
23
24
25
26
27
def get_all(self, active_only: bool = False) -> list[SpecialEvent]:
    """Get all special events, optionally filtered by active status.

    Args:
        active_only: If True, only return active events. Defaults to False.

    Returns:
        List of SpecialEvent instances ordered by name.
    """
    query = self.session.query(SpecialEvent).order_by(SpecialEvent.name)
    if active_only:
        query = query.filter_by(active=True)
    return query.all()

get_active()

Get all active special events.

Convenience method for getting active events only.

Returns:

Type Description
list[SpecialEvent]

List of active SpecialEvent instances ordered by name.
Source code in website/repositories/special_event.py 
29
30
31
32
33
34
35
36
37
def get_active(self) -> list[SpecialEvent]:
    """Get all active special events.

    Convenience method for getting active events only.

    Returns:
        List of active SpecialEvent instances ordered by name.
    """
    return self.get_all(active_only=True)

get_by_name(name)

Get special event by name.

Parameters:

Name Type Description Default
name str

Name of the special event.

 required

Returns:

Type Description
SpecialEvent | None

SpecialEvent instance if found, None otherwise.
Source code in website/repositories/special_event.py 
39
40
41
42
43
44
45
46
47
48
def get_by_name(self, name: str) -> SpecialEvent | None:
    """Get special event by name.

    Args:
        name: Name of the special event.

    Returns:
        SpecialEvent instance if found, None otherwise.
    """
    return self.session.query(SpecialEvent).filter_by(name=name).first()

SystemRepository

Bases: BaseRepository[System]

Repository for System entities.

Source code in website/repositories/system.py 
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
class SystemRepository(BaseRepository[System]):
    """Repository for System entities."""

    model_class = System

    def get_all_ordered(self) -> list[System]:
        """Retrieve all systems ordered by name.

        Returns:
            List of System instances sorted alphabetically.
        """
        return self.session.query(System).order_by(System.name).all()

    def get_by_name(self, name: str) -> System | None:
        """Find a system by its name.

        Args:
            name: System name to search for.

        Returns:
            System instance if found, None otherwise.
        """
        return self.session.query(System).filter_by(name=name).first()

get_all_ordered()

Retrieve all systems ordered by name.

Returns:

Type Description
list[System]

List of System instances sorted alphabetically.
Source code in website/repositories/system.py 
12
13
14
15
16
17
18
def get_all_ordered(self) -> list[System]:
    """Retrieve all systems ordered by name.

    Returns:
        List of System instances sorted alphabetically.
    """
    return self.session.query(System).order_by(System.name).all()

get_by_name(name)

Find a system by its name.

Parameters:

Name Type Description Default
name str

System name to search for.

 required

Returns:

Type Description
System | None

System instance if found, None otherwise.
Source code in website/repositories/system.py 
20
21
22
23
24
25
26
27
28
29
def get_by_name(self, name: str) -> System | None:
    """Find a system by its name.

    Args:
        name: System name to search for.

    Returns:
        System instance if found, None otherwise.
    """
    return self.session.query(System).filter_by(name=name).first()

TrophyRepository

Bases: BaseRepository[Trophy]

Repository for Trophy data access.

Handles querying trophies, user trophies, and leaderboard aggregations.

Source code in website/repositories/trophy.py 
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
class TrophyRepository(BaseRepository[Trophy]):
    """Repository for Trophy data access.

    Handles querying trophies, user trophies, and leaderboard aggregations.
    """

    model_class = Trophy

    def get_user_trophy(self, user_id: str, trophy_id: int) -> UserTrophy | None:
        """Get a user's trophy record.

        Args:
            user_id: User ID.
            trophy_id: Trophy ID.

        Returns:
            UserTrophy instance if user has this trophy, None otherwise.
        """
        return (
            self.session.query(UserTrophy).filter_by(user_id=user_id, trophy_id=trophy_id).first()
        )

    def award_trophy(self, user_id: str, trophy_id: int, amount: int = 1) -> UserTrophy:
        """Create or update a user trophy record.

        Args:
            user_id: User ID.
            trophy_id: Trophy ID.
            amount: Quantity to add. Defaults to 1.

        Returns:
            Created or updated UserTrophy instance.
        """
        user_trophy = self.get_user_trophy(user_id, trophy_id)

        if user_trophy:
            user_trophy.quantity += amount
        else:
            user_trophy = UserTrophy(user_id=user_id, trophy_id=trophy_id, quantity=amount)
            self.session.add(user_trophy)

        self.session.flush()
        return user_trophy

    def get_leaderboard(self, trophy_id: int, limit: int = 10) -> list[tuple[User, int]]:
        """Get leaderboard for a specific trophy.

        Args:
            trophy_id: Trophy ID to get leaderboard for.
            limit: Maximum number of entries to return. Defaults to 10.

        Returns:
            List of (User, total_quantity) tuples ordered by quantity descending.
        """
        return (
            self.session.query(User, func.sum(UserTrophy.quantity).label("total"))
            .join(UserTrophy)
            .filter(UserTrophy.trophy_id == trophy_id)
            .group_by(User.id)
            .order_by(func.sum(UserTrophy.quantity).desc())
            .limit(limit)
            .all()
        )

get_user_trophy(user_id, trophy_id)

Get a user's trophy record.

Parameters:

Name Type Description Default
user_id str

User ID.

 required
trophy_id int

Trophy ID.

 required

Returns:

Type Description
UserTrophy | None

UserTrophy instance if user has this trophy, None otherwise.
Source code in website/repositories/trophy.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
def get_user_trophy(self, user_id: str, trophy_id: int) -> UserTrophy | None:
    """Get a user's trophy record.

    Args:
        user_id: User ID.
        trophy_id: Trophy ID.

    Returns:
        UserTrophy instance if user has this trophy, None otherwise.
    """
    return (
        self.session.query(UserTrophy).filter_by(user_id=user_id, trophy_id=trophy_id).first()
    )

award_trophy(user_id, trophy_id, amount=1)

Create or update a user trophy record.

Parameters:

Name Type Description Default
user_id str

User ID.

 required
trophy_id int

Trophy ID.

 required
amount int

Quantity to add. Defaults to 1.

 1

Returns:

Type Description
UserTrophy

Created or updated UserTrophy instance.
Source code in website/repositories/trophy.py 
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
def award_trophy(self, user_id: str, trophy_id: int, amount: int = 1) -> UserTrophy:
    """Create or update a user trophy record.

    Args:
        user_id: User ID.
        trophy_id: Trophy ID.
        amount: Quantity to add. Defaults to 1.

    Returns:
        Created or updated UserTrophy instance.
    """
    user_trophy = self.get_user_trophy(user_id, trophy_id)

    if user_trophy:
        user_trophy.quantity += amount
    else:
        user_trophy = UserTrophy(user_id=user_id, trophy_id=trophy_id, quantity=amount)
        self.session.add(user_trophy)

    self.session.flush()
    return user_trophy

get_leaderboard(trophy_id, limit=10)

Get leaderboard for a specific trophy.

Parameters:

Name Type Description Default
trophy_id int

Trophy ID to get leaderboard for.

 required
limit int

Maximum number of entries to return. Defaults to 10.

 10

Returns:

Type Description
list[tuple[User, int]]

List of (User, total_quantity) tuples ordered by quantity descending.
Source code in website/repositories/trophy.py 
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
def get_leaderboard(self, trophy_id: int, limit: int = 10) -> list[tuple[User, int]]:
    """Get leaderboard for a specific trophy.

    Args:
        trophy_id: Trophy ID to get leaderboard for.
        limit: Maximum number of entries to return. Defaults to 10.

    Returns:
        List of (User, total_quantity) tuples ordered by quantity descending.
    """
    return (
        self.session.query(User, func.sum(UserTrophy.quantity).label("total"))
        .join(UserTrophy)
        .filter(UserTrophy.trophy_id == trophy_id)
        .group_by(User.id)
        .order_by(func.sum(UserTrophy.quantity).desc())
        .limit(limit)
        .all()
    )

UserRepository

Bases: BaseRepository[User]

Repository for User entities.

Source code in website/repositories/user.py 
 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
class UserRepository(BaseRepository[User]):
    """Repository for User entities."""

    model_class = User

    def get_active_users(self) -> list[User]:
        """Retrieve all users not marked as inactive.

        Returns:
            List of User instances where not_player_as_of is NULL.
        """
        return self.session.query(User).filter(User.not_player_as_of.is_(None)).all()

    def get_active_user_ids(self) -> list[str]:
        """Retrieve IDs of all users not marked as inactive.

        Uses a scalar query to avoid loading full ORM objects
        (and triggering init_on_load).

        Returns:
            List of user ID strings where not_player_as_of is NULL.
        """
        rows = self.session.query(User.id).filter(User.not_player_as_of.is_(None)).all()
        return [row[0] for row in rows]

    def get_inactive_user_ids(self) -> list[str]:
        """Retrieve IDs of all users marked as inactive.

        Uses a scalar query to avoid loading full ORM objects
        (and triggering init_on_load).

        Returns:
            List of user ID strings where not_player_as_of is set.
        """
        rows = self.session.query(User.id).filter(User.not_player_as_of.isnot(None)).all()
        return [row[0] for row in rows]

    def get_by_ids(self, ids: list[str]) -> list[User]:
        """Retrieve users by a list of IDs.

        Args:
            ids: List of user ID strings.

        Returns:
            List of User instances matching the given IDs.
        """
        if not ids:
            return []
        return self.session.query(User).filter(User.id.in_(ids)).all()

    def get_inactive_users(self) -> list[User]:
        """Retrieve all users marked as inactive.

        Returns:
            List of User instances where not_player_as_of is set.
        """
        return self.session.query(User).filter(User.not_player_as_of.isnot(None)).all()

get_active_users()

Retrieve all users not marked as inactive.

Returns:

Type Description
list[User]

List of User instances where not_player_as_of is NULL.
Source code in website/repositories/user.py 
12
13
14
15
16
17
18
def get_active_users(self) -> list[User]:
    """Retrieve all users not marked as inactive.

    Returns:
        List of User instances where not_player_as_of is NULL.
    """
    return self.session.query(User).filter(User.not_player_as_of.is_(None)).all()

get_active_user_ids()

Retrieve IDs of all users not marked as inactive.

Uses a scalar query to avoid loading full ORM objects (and triggering init_on_load).

Returns:

Type Description
list[str]

List of user ID strings where not_player_as_of is NULL.
Source code in website/repositories/user.py 
20
21
22
23
24
25
26
27
28
29
30
def get_active_user_ids(self) -> list[str]:
    """Retrieve IDs of all users not marked as inactive.

    Uses a scalar query to avoid loading full ORM objects
    (and triggering init_on_load).

    Returns:
        List of user ID strings where not_player_as_of is NULL.
    """
    rows = self.session.query(User.id).filter(User.not_player_as_of.is_(None)).all()
    return [row[0] for row in rows]

get_inactive_user_ids()

Retrieve IDs of all users marked as inactive.

Uses a scalar query to avoid loading full ORM objects (and triggering init_on_load).

Returns:

Type Description
list[str]

List of user ID strings where not_player_as_of is set.
Source code in website/repositories/user.py 
32
33
34
35
36
37
38
39
40
41
42
def get_inactive_user_ids(self) -> list[str]:
    """Retrieve IDs of all users marked as inactive.

    Uses a scalar query to avoid loading full ORM objects
    (and triggering init_on_load).

    Returns:
        List of user ID strings where not_player_as_of is set.
    """
    rows = self.session.query(User.id).filter(User.not_player_as_of.isnot(None)).all()
    return [row[0] for row in rows]

get_by_ids(ids)

Retrieve users by a list of IDs.

Parameters:

Name Type Description Default
ids list[str]

List of user ID strings.

 required

Returns:

Type Description
list[User]

List of User instances matching the given IDs.
Source code in website/repositories/user.py 
44
45
46
47
48
49
50
51
52
53
54
55
def get_by_ids(self, ids: list[str]) -> list[User]:
    """Retrieve users by a list of IDs.

    Args:
        ids: List of user ID strings.

    Returns:
        List of User instances matching the given IDs.
    """
    if not ids:
        return []
    return self.session.query(User).filter(User.id.in_(ids)).all()

get_inactive_users()

Retrieve all users marked as inactive.

Returns:

Type Description
list[User]

List of User instances where not_player_as_of is set.
Source code in website/repositories/user.py 
57
58
59
60
61
62
63
def get_inactive_users(self) -> list[User]:
    """Retrieve all users marked as inactive.

    Returns:
        List of User instances where not_player_as_of is set.
    """
    return self.session.query(User).filter(User.not_player_as_of.isnot(None)).all()

VttRepository

Bases: BaseRepository[Vtt]

Repository for Vtt entities.

Source code in website/repositories/vtt.py 
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
class VttRepository(BaseRepository[Vtt]):
    """Repository for Vtt entities."""

    model_class = Vtt

    def get_all_ordered(self) -> list[Vtt]:
        """Retrieve all VTTs ordered by name.

        Returns:
            List of Vtt instances sorted alphabetically.
        """
        return self.session.query(Vtt).order_by(Vtt.name).all()

    def get_by_name(self, name: str) -> Vtt | None:
        """Find a VTT by its name.

        Args:
            name: VTT name to search for.

        Returns:
            Vtt instance if found, None otherwise.
        """
        return self.session.query(Vtt).filter_by(name=name).first()

get_all_ordered()

Retrieve all VTTs ordered by name.

Returns:

Type Description
list[Vtt]

List of Vtt instances sorted alphabetically.
Source code in website/repositories/vtt.py 
12
13
14
15
16
17
18
def get_all_ordered(self) -> list[Vtt]:
    """Retrieve all VTTs ordered by name.

    Returns:
        List of Vtt instances sorted alphabetically.
    """
    return self.session.query(Vtt).order_by(Vtt.name).all()

get_by_name(name)

Find a VTT by its name.

Parameters:

Name Type Description Default
name str

VTT name to search for.

 required

Returns:

Type Description
Vtt | None

Vtt instance if found, None otherwise.
Source code in website/repositories/vtt.py 
20
21
22
23
24
25
26
27
28
29
def get_by_name(self, name: str) -> Vtt | None:
    """Find a VTT by its name.

    Args:
        name: VTT name to search for.

    Returns:
        Vtt instance if found, None otherwise.
    """
    return self.session.query(Vtt).filter_by(name=name).first()