Services

The service layer lives in website/services/ and contains all business logic. Services are the primary place for new logic in QuestMaster.

Services:

• Own transaction boundaries
• Perform validation and enforce business rules
• Raise domain-specific exceptions from website.exceptions
• Call repositories for data access
• Never access Flask request or session directly

Overview

Service	Repository	Model	Description
ChannelService	ChannelRepository	Channel	Category size management and Discord channel cleanup
DiscordService	Discord (client)	—	Discord API wrapper with dependency injection for testability
GameService	GameRepository	Game	Complete game lifecycle — creation, publishing, registration, archival, Discord sync
GameEventService	GameEventRepository	GameEvent	Transaction-safe audit trail logging for games
GameSessionService	GameSessionRepository	GameSession	Session CRUD with conflict detection and validation
SpecialEventService	SpecialEventRepository	SpecialEvent	Special event CRUD with uniqueness validation
SystemService	SystemRepository	System	Game system CRUD with cache invalidation
TrophyService	TrophyRepository	Trophy	Trophy awarding logic (unique vs. non-unique rules) and leaderboards
UserService	UserRepository	User	User retrieval, creation, and Discord profile initialization
VttService	VttRepository	Vtt	Virtual tabletop CRUD with cache invalidation

API Reference

Service layer for business logic and transaction management.

ChannelService

Service layer for Channel (Discord category) management.

Handles category size tracking for Discord channel organization.

Source code in website/services/channel.py

class ChannelService:
    """Service layer for Channel (Discord category) management.

    Handles category size tracking for Discord channel organization.
    """

    def __init__(self, repository=None):
        self.repo = repository or ChannelRepository()

    def get_category(self, game_type: str) -> Channel:
        """Get the smallest category for a game type.

        Args:
            game_type: Type of game (oneshot, campaign).

        Returns:
            Channel category with smallest size.

        Raises:
            NotFoundError: If no category found for type.
        """
        category = self.repo.get_smallest_by_type(game_type)
        if not category:
            raise NotFoundError(
                f"No channel category found for type '{game_type}'",
                resource_type="Channel",
            )
        return category

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

        Args:
            channel: Channel category to increment.
        """
        self.repo.increment_size(channel)

    def adjust_category_size(self, discord_service: DiscordService, game: Game) -> None:
        """Decrement category size when a game channel is deleted.

        Args:
            discord_service: DiscordService instance for API calls.
            game: Game instance with channel to look up.
        """
        try:
            discord_channel = discord_service.get_channel(game.channel)
            parent_id = discord_channel.get("parent_id")
            if parent_id:
                category = self.repo.get_by_id(parent_id)
                if category:
                    self.repo.decrement_size(category)
                    db.session.commit()
                    logger.info(f"Decreased size of category {category.id} to {category.size}")
        except Exception as e:
            logger.warning(f"Failed to adjust category size for game {game.id}: {e}")

get_category(game_type)

Get the smallest category for a game type.

Parameters:

Name	Type	Description	Default
game_type	str	Type of game (oneshot, campaign).	required

Returns:

Type	Description
Channel	Channel category with smallest size.

Raises:

Type	Description
NotFoundError	If no category found for type.

Source code in website/services/channel.py

def get_category(self, game_type: str) -> Channel:
    """Get the smallest category for a game type.

    Args:
        game_type: Type of game (oneshot, campaign).

    Returns:
        Channel category with smallest size.

    Raises:
        NotFoundError: If no category found for type.
    """
    category = self.repo.get_smallest_by_type(game_type)
    if not category:
        raise NotFoundError(
            f"No channel category found for type '{game_type}'",
            resource_type="Channel",
        )
    return category

increment_size(channel)

Increment the channel count for a category.

Parameters:

Name	Type	Description	Default
channel	Channel	Channel category to increment.	required

Source code in website/services/channel.py

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

    Args:
        channel: Channel category to increment.
    """
    self.repo.increment_size(channel)

adjust_category_size(discord_service, game)

Decrement category size when a game channel is deleted.

Parameters:

Name	Type	Description	Default
discord_service	DiscordService	DiscordService instance for API calls.	required
game	Game	Game instance with channel to look up.	required

Source code in website/services/channel.py

def adjust_category_size(self, discord_service: DiscordService, game: Game) -> None:
    """Decrement category size when a game channel is deleted.

    Args:
        discord_service: DiscordService instance for API calls.
        game: Game instance with channel to look up.
    """
    try:
        discord_channel = discord_service.get_channel(game.channel)
        parent_id = discord_channel.get("parent_id")
        if parent_id:
            category = self.repo.get_by_id(parent_id)
            if category:
                self.repo.decrement_size(category)
                db.session.commit()
                logger.info(f"Decreased size of category {category.id} to {category.size}")
    except Exception as e:
        logger.warning(f"Failed to adjust category size for game {game.id}: {e}")

DiscordService

Service layer for Discord API interactions.

Provides a testable wrapper around the Discord API client. Uses dependency injection to allow mocking in tests.

Attributes:

Name	Type	Description
bot	Discord	The underlying Discord API client instance.

Source code in website/services/discord.py

class DiscordService:
    """Service layer for Discord API interactions.

    Provides a testable wrapper around the Discord API client. Uses dependency
    injection to allow mocking in tests.

    Attributes:
        bot: The underlying Discord API client instance.
    """

    def __init__(self, bot: Optional[Discord] = None):
        self._bot = bot

    @property
    def bot(self) -> Discord:
        """Get the Discord bot instance.

        Lazy-loads from the global singleton if not injected.

        Returns:
            Discord client instance.

        Raises:
            RuntimeError: If no bot instance is available.
        """
        if self._bot is None:
            from website.bot import get_bot

            self._bot = get_bot()
        if self._bot is None:
            raise RuntimeError("Discord bot not initialized")
        return self._bot

    # -------------------------------------------------------------------------
    # User operations
    # -------------------------------------------------------------------------

    def get_user(self, user_id: str) -> dict:
        """Fetch guild member data from Discord.

        Args:
            user_id: Discord user ID.

        Returns:
            Member data dictionary from Discord API.

        Raises:
            DiscordAPIError: If the API request fails.
        """
        return self.bot.get_user(user_id)

    def add_role_to_user(self, user_id: str, role_id: str) -> dict:
        """Add a role to a user.

        Args:
            user_id: Discord user ID.
            role_id: Discord role ID.

        Returns:
            API response (usually empty on success).

        Raises:
            DiscordAPIError: If the API request fails.
        """
        return self.bot.add_role_to_user(user_id, role_id)

    def remove_role_from_user(self, user_id: str, role_id: str) -> dict:
        """Remove a role from a user.

        Args:
            user_id: Discord user ID.
            role_id: Discord role ID.

        Returns:
            API response (usually empty on success).

        Raises:
            DiscordAPIError: If the API request fails.
        """
        return self.bot.remove_role_from_user(user_id, role_id)

    # -------------------------------------------------------------------------
    # Role operations
    # -------------------------------------------------------------------------

    def create_role(
        self,
        name: str,
        permissions: str = PLAYER_ROLE_PERMISSION,
        color: int = 0,
    ) -> dict:
        """Create a Discord role.

        Args:
            name: Role name (will be sanitized).
            permissions: Permission bitfield string.
            color: Role color as integer.

        Returns:
            Created role data including 'id'.

        Raises:
            DiscordAPIError: If the API request fails.
        """
        return self.bot.create_role(name, permissions, color)

    def get_role(self, role_id: str) -> dict:
        """Get a role by ID.

        Args:
            role_id: Discord role ID.

        Returns:
            Role data dictionary.

        Raises:
            DiscordAPIError: If the API request fails.
        """
        return self.bot.get_role(role_id)

    def delete_role(self, role_id: str) -> dict:
        """Delete a Discord role.

        Args:
            role_id: Discord role ID.

        Returns:
            API response (usually empty on success).

        Raises:
            DiscordAPIError: If the API request fails.
        """
        return self.bot.delete_role(role_id)

    # -------------------------------------------------------------------------
    # Channel operations
    # -------------------------------------------------------------------------

    def create_channel(
        self,
        name: str,
        parent_id: str,
        role_id: str,
        gm_id: str,
    ) -> dict:
        """Create a Discord text channel with permissions.

        Args:
            name: Channel name (will be sanitized).
            parent_id: Parent category ID.
            role_id: Player role ID for permission overwrites.
            gm_id: GM user ID for permission overwrites.

        Returns:
            Created channel data including 'id'.

        Raises:
            DiscordAPIError: If the API request fails.
        """
        return self.bot.create_channel(name, parent_id, role_id, gm_id)

    def get_channel(self, channel_id: str) -> dict:
        """Get a channel by ID.

        Args:
            channel_id: Discord channel ID.

        Returns:
            Channel data dictionary.

        Raises:
            DiscordAPIError: If the API request fails.
        """
        return self.bot.get_channel(channel_id)

    def delete_channel(self, channel_id: str) -> dict:
        """Delete a Discord channel.

        Args:
            channel_id: Discord channel ID.

        Returns:
            API response (usually empty on success).

        Raises:
            DiscordAPIError: If the API request fails.
        """
        return self.bot.delete_channel(channel_id)

    # -------------------------------------------------------------------------
    # Message operations
    # -------------------------------------------------------------------------

    def send_message(self, content: str, channel_id: str) -> dict:
        """Send a plain text message to a channel.

        Args:
            content: Message content.
            channel_id: Target channel ID.

        Returns:
            Created message data including 'id'.

        Raises:
            DiscordAPIError: If the API request fails.
        """
        return self.bot.send_message(content, channel_id)

    def delete_message(self, message_id: str, channel_id: str) -> dict:
        """Delete a message.

        Args:
            message_id: Discord message ID.
            channel_id: Channel containing the message.

        Returns:
            API response (usually empty on success).

        Raises:
            DiscordAPIError: If the API request fails.
        """
        return self.bot.delete_message(message_id, channel_id)

    def send_embed(self, embed: dict, channel_id: str) -> dict:
        """Send an embed message to a channel.

        Args:
            embed: Embed data dictionary.
            channel_id: Target channel ID.

        Returns:
            Created message data including 'id'.

        Raises:
            DiscordAPIError: If the API request fails.
        """
        return self.bot.send_embed_message(embed, channel_id)

    def edit_embed(self, message_id: str, embed: dict, channel_id: str) -> dict:
        """Edit an existing embed message.

        Args:
            message_id: Message ID to edit.
            embed: New embed data dictionary.
            channel_id: Channel containing the message.

        Returns:
            Updated message data.

        Raises:
            DiscordAPIError: If the API request fails.
        """
        return self.bot.edit_embed_message(message_id, embed, channel_id)

    def pin_message(self, message_id: str, channel_id: str) -> dict:
        """Pin a message to a channel.

        Args:
            message_id: Message ID to edit.
            channel_id: Target channel ID.

        Returns:
            API response (usually empty on success).

        Raises:
            DiscordAPIError: If the API request fails.
        """
        return self.bot.pin_message(message_id, channel_id)

    # -------------------------------------------------------------------------
    # Game embed operations (high-level)
    # -------------------------------------------------------------------------

    def send_game_embed(
        self,
        game: Game,
        embed_type: str = "annonce",
        start: Optional[datetime] = None,
        end: Optional[datetime] = None,
        player: Optional[str] = None,
        old_start: Optional[datetime] = None,
        old_end: Optional[datetime] = None,
        alert_message: Optional[str] = None,
    ) -> str:
        """Send or update a Discord embed for a game event.

        This is a high-level method that builds and sends the appropriate embed
        based on the embed_type.

        Args:
            game: Game model instance.
            embed_type: Type of embed ('annonce', 'annonce_details', 'add-session',
                'edit-session', 'del-session', 'register', 'alert').
            start: Session start datetime (for session embeds).
            end: Session end datetime (for session embeds).
            player: Player user ID (for register/alert embeds).
            old_start: Previous session start (for edit-session).
            old_end: Previous session end (for edit-session).
            alert_message: Alert text (for alert embed).

        Returns:
            Discord message ID string.

        Raises:
            ValueError: If embed_type is unknown.
            DiscordAPIError: If the API request fails.
        """
        from website.utils.game_embeds import (
            build_add_session_embed,
            build_alert_embed,
            build_annonce_details_embed,
            build_annonce_embed,
            build_delete_session_embed,
            build_edit_session_embed,
            build_register_embed,
        )

        embed_builders = {
            "annonce": build_annonce_embed,
            "annonce_details": build_annonce_details_embed,
            "add-session": build_add_session_embed,
            "edit-session": build_edit_session_embed,
            "del-session": build_delete_session_embed,
            "register": build_register_embed,
            "alert": build_alert_embed,
        }

        if embed_type not in embed_builders:
            raise ValueError(f"Unknown embed type: {embed_type}")

        embed, target = embed_builders[embed_type](
            game, start, end, player, old_start, old_end, alert_message
        )

        if embed_type == "annonce" and game.msg_id:
            response = self.edit_embed(game.msg_id, embed, target)
        else:
            response = self.send_embed(embed, target)

        return response["id"]

bot property

Get the Discord bot instance.

Lazy-loads from the global singleton if not injected.

Returns:

Type	Description
Discord	Discord client instance.

Raises:

Type	Description
RuntimeError	If no bot instance is available.

get_user(user_id)

Fetch guild member data from Discord.

Parameters:

Name	Type	Description	Default
user_id	str	Discord user ID.	required

Returns:

Type	Description
dict	Member data dictionary from Discord API.

Raises:

Type	Description
DiscordAPIError	If the API request fails.

Source code in website/services/discord.py

def get_user(self, user_id: str) -> dict:
    """Fetch guild member data from Discord.

    Args:
        user_id: Discord user ID.

    Returns:
        Member data dictionary from Discord API.

    Raises:
        DiscordAPIError: If the API request fails.
    """
    return self.bot.get_user(user_id)

add_role_to_user(user_id, role_id)

Add a role to a user.

Parameters:

Name	Type	Description	Default
user_id	str	Discord user ID.	required
role_id	str	Discord role ID.	required

Returns:

Type	Description
dict	API response (usually empty on success).

Raises:

Type	Description
DiscordAPIError	If the API request fails.

Source code in website/services/discord.py

def add_role_to_user(self, user_id: str, role_id: str) -> dict:
    """Add a role to a user.

    Args:
        user_id: Discord user ID.
        role_id: Discord role ID.

    Returns:
        API response (usually empty on success).

    Raises:
        DiscordAPIError: If the API request fails.
    """
    return self.bot.add_role_to_user(user_id, role_id)

remove_role_from_user(user_id, role_id)

Remove a role from a user.

Parameters:

Name	Type	Description	Default
user_id	str	Discord user ID.	required
role_id	str	Discord role ID.	required

Returns:

Type	Description
dict	API response (usually empty on success).

Raises:

Type	Description
DiscordAPIError	If the API request fails.

Source code in website/services/discord.py

def remove_role_from_user(self, user_id: str, role_id: str) -> dict:
    """Remove a role from a user.

    Args:
        user_id: Discord user ID.
        role_id: Discord role ID.

    Returns:
        API response (usually empty on success).

    Raises:
        DiscordAPIError: If the API request fails.
    """
    return self.bot.remove_role_from_user(user_id, role_id)

create_role(name, permissions=PLAYER_ROLE_PERMISSION, color=0)

Create a Discord role.

Parameters:

Name	Type	Description	Default
name	str	Role name (will be sanitized).	required
permissions	str	Permission bitfield string.	PLAYER_ROLE_PERMISSION
color	int	Role color as integer.	0

Returns:

Type	Description
dict	Created role data including 'id'.

Raises:

Type	Description
DiscordAPIError	If the API request fails.

Source code in website/services/discord.py

def create_role(
    self,
    name: str,
    permissions: str = PLAYER_ROLE_PERMISSION,
    color: int = 0,
) -> dict:
    """Create a Discord role.

    Args:
        name: Role name (will be sanitized).
        permissions: Permission bitfield string.
        color: Role color as integer.

    Returns:
        Created role data including 'id'.

    Raises:
        DiscordAPIError: If the API request fails.
    """
    return self.bot.create_role(name, permissions, color)

get_role(role_id)

Get a role by ID.

Parameters:

Name	Type	Description	Default
role_id	str	Discord role ID.	required

Returns:

Type	Description
dict	Role data dictionary.

Raises:

Type	Description
DiscordAPIError	If the API request fails.

Source code in website/services/discord.py

def get_role(self, role_id: str) -> dict:
    """Get a role by ID.

    Args:
        role_id: Discord role ID.

    Returns:
        Role data dictionary.

    Raises:
        DiscordAPIError: If the API request fails.
    """
    return self.bot.get_role(role_id)

delete_role(role_id)

Delete a Discord role.

Parameters:

Name	Type	Description	Default
role_id	str	Discord role ID.	required

Returns:

Type	Description
dict	API response (usually empty on success).

Raises:

Type	Description
DiscordAPIError	If the API request fails.

Source code in website/services/discord.py

def delete_role(self, role_id: str) -> dict:
    """Delete a Discord role.

    Args:
        role_id: Discord role ID.

    Returns:
        API response (usually empty on success).

    Raises:
        DiscordAPIError: If the API request fails.
    """
    return self.bot.delete_role(role_id)

create_channel(name, parent_id, role_id, gm_id)

Create a Discord text channel with permissions.

Parameters:

Name	Type	Description	Default
name	str	Channel name (will be sanitized).	required
parent_id	str	Parent category ID.	required
role_id	str	Player role ID for permission overwrites.	required
gm_id	str	GM user ID for permission overwrites.	required

Returns:

Type	Description
dict	Created channel data including 'id'.

Raises:

Type	Description
DiscordAPIError	If the API request fails.

Source code in website/services/discord.py

def create_channel(
    self,
    name: str,
    parent_id: str,
    role_id: str,
    gm_id: str,
) -> dict:
    """Create a Discord text channel with permissions.

    Args:
        name: Channel name (will be sanitized).
        parent_id: Parent category ID.
        role_id: Player role ID for permission overwrites.
        gm_id: GM user ID for permission overwrites.

    Returns:
        Created channel data including 'id'.

    Raises:
        DiscordAPIError: If the API request fails.
    """
    return self.bot.create_channel(name, parent_id, role_id, gm_id)

get_channel(channel_id)

Get a channel by ID.

Parameters:

Name	Type	Description	Default
channel_id	str	Discord channel ID.	required

Returns:

Type	Description
dict	Channel data dictionary.

Raises:

Type	Description
DiscordAPIError	If the API request fails.

Source code in website/services/discord.py

def get_channel(self, channel_id: str) -> dict:
    """Get a channel by ID.

    Args:
        channel_id: Discord channel ID.

    Returns:
        Channel data dictionary.

    Raises:
        DiscordAPIError: If the API request fails.
    """
    return self.bot.get_channel(channel_id)

delete_channel(channel_id)

Delete a Discord channel.

Parameters:

Name	Type	Description	Default
channel_id	str	Discord channel ID.	required

Returns:

Type	Description
dict	API response (usually empty on success).

Raises:

Type	Description
DiscordAPIError	If the API request fails.

Source code in website/services/discord.py

def delete_channel(self, channel_id: str) -> dict:
    """Delete a Discord channel.

    Args:
        channel_id: Discord channel ID.

    Returns:
        API response (usually empty on success).

    Raises:
        DiscordAPIError: If the API request fails.
    """
    return self.bot.delete_channel(channel_id)

send_message(content, channel_id)

Send a plain text message to a channel.

Parameters:

Name	Type	Description	Default
content	str	Message content.	required
channel_id	str	Target channel ID.	required

Returns:

Type	Description
dict	Created message data including 'id'.

Raises:

Type	Description
DiscordAPIError	If the API request fails.

Source code in website/services/discord.py

def send_message(self, content: str, channel_id: str) -> dict:
    """Send a plain text message to a channel.

    Args:
        content: Message content.
        channel_id: Target channel ID.

    Returns:
        Created message data including 'id'.

    Raises:
        DiscordAPIError: If the API request fails.
    """
    return self.bot.send_message(content, channel_id)

delete_message(message_id, channel_id)

Delete a message.

Parameters:

Name	Type	Description	Default
message_id	str	Discord message ID.	required
channel_id	str	Channel containing the message.	required

Returns:

Type	Description
dict	API response (usually empty on success).

Raises:

Type	Description
DiscordAPIError	If the API request fails.

Source code in website/services/discord.py

def delete_message(self, message_id: str, channel_id: str) -> dict:
    """Delete a message.

    Args:
        message_id: Discord message ID.
        channel_id: Channel containing the message.

    Returns:
        API response (usually empty on success).

    Raises:
        DiscordAPIError: If the API request fails.
    """
    return self.bot.delete_message(message_id, channel_id)

send_embed(embed, channel_id)

Send an embed message to a channel.

Parameters:

Name	Type	Description	Default
embed	dict	Embed data dictionary.	required
channel_id	str	Target channel ID.	required

Returns:

Type	Description
dict	Created message data including 'id'.

Raises:

Type	Description
DiscordAPIError	If the API request fails.

Source code in website/services/discord.py

def send_embed(self, embed: dict, channel_id: str) -> dict:
    """Send an embed message to a channel.

    Args:
        embed: Embed data dictionary.
        channel_id: Target channel ID.

    Returns:
        Created message data including 'id'.

    Raises:
        DiscordAPIError: If the API request fails.
    """
    return self.bot.send_embed_message(embed, channel_id)

edit_embed(message_id, embed, channel_id)

Edit an existing embed message.

Parameters:

Name	Type	Description	Default
message_id	str	Message ID to edit.	required
embed	dict	New embed data dictionary.	required
channel_id	str	Channel containing the message.	required

Returns:

Type	Description
dict	Updated message data.

Raises:

Type	Description
DiscordAPIError	If the API request fails.

Source code in website/services/discord.py

def edit_embed(self, message_id: str, embed: dict, channel_id: str) -> dict:
    """Edit an existing embed message.

    Args:
        message_id: Message ID to edit.
        embed: New embed data dictionary.
        channel_id: Channel containing the message.

    Returns:
        Updated message data.

    Raises:
        DiscordAPIError: If the API request fails.
    """
    return self.bot.edit_embed_message(message_id, embed, channel_id)

pin_message(message_id, channel_id)

Pin a message to a channel.

Parameters:

Name	Type	Description	Default
message_id	str	Message ID to edit.	required
channel_id	str	Target channel ID.	required

Returns:

Type	Description
dict	API response (usually empty on success).

Raises:

Type	Description
DiscordAPIError	If the API request fails.

Source code in website/services/discord.py

def pin_message(self, message_id: str, channel_id: str) -> dict:
    """Pin a message to a channel.

    Args:
        message_id: Message ID to edit.
        channel_id: Target channel ID.

    Returns:
        API response (usually empty on success).

    Raises:
        DiscordAPIError: If the API request fails.
    """
    return self.bot.pin_message(message_id, channel_id)

send_game_embed(game, embed_type='annonce', start=None, end=None, player=None, old_start=None, old_end=None, alert_message=None)

Send or update a Discord embed for a game event.

This is a high-level method that builds and sends the appropriate embed based on the embed_type.

Parameters:

Name	Type	Description	Default
game	Game	Game model instance.	required
embed_type	str	Type of embed ('annonce', 'annonce_details', 'add-session', 'edit-session', 'del-session', 'register', 'alert').	'annonce'
start	Optional[datetime]	Session start datetime (for session embeds).	None
end	Optional[datetime]	Session end datetime (for session embeds).	None
player	Optional[str]	Player user ID (for register/alert embeds).	None
old_start	Optional[datetime]	Previous session start (for edit-session).	None
old_end	Optional[datetime]	Previous session end (for edit-session).	None
alert_message	Optional[str]	Alert text (for alert embed).	None

Returns:

Type	Description
str	Discord message ID string.

Raises:

Type	Description
ValueError	If embed_type is unknown.
DiscordAPIError	If the API request fails.

Source code in website/services/discord.py

def send_game_embed(
    self,
    game: Game,
    embed_type: str = "annonce",
    start: Optional[datetime] = None,
    end: Optional[datetime] = None,
    player: Optional[str] = None,
    old_start: Optional[datetime] = None,
    old_end: Optional[datetime] = None,
    alert_message: Optional[str] = None,
) -> str:
    """Send or update a Discord embed for a game event.

    This is a high-level method that builds and sends the appropriate embed
    based on the embed_type.

    Args:
        game: Game model instance.
        embed_type: Type of embed ('annonce', 'annonce_details', 'add-session',
            'edit-session', 'del-session', 'register', 'alert').
        start: Session start datetime (for session embeds).
        end: Session end datetime (for session embeds).
        player: Player user ID (for register/alert embeds).
        old_start: Previous session start (for edit-session).
        old_end: Previous session end (for edit-session).
        alert_message: Alert text (for alert embed).

    Returns:
        Discord message ID string.

    Raises:
        ValueError: If embed_type is unknown.
        DiscordAPIError: If the API request fails.
    """
    from website.utils.game_embeds import (
        build_add_session_embed,
        build_alert_embed,
        build_annonce_details_embed,
        build_annonce_embed,
        build_delete_session_embed,
        build_edit_session_embed,
        build_register_embed,
    )

    embed_builders = {
        "annonce": build_annonce_embed,
        "annonce_details": build_annonce_details_embed,
        "add-session": build_add_session_embed,
        "edit-session": build_edit_session_embed,
        "del-session": build_delete_session_embed,
        "register": build_register_embed,
        "alert": build_alert_embed,
    }

    if embed_type not in embed_builders:
        raise ValueError(f"Unknown embed type: {embed_type}")

    embed, target = embed_builders[embed_type](
        game, start, end, player, old_start, old_end, alert_message
    )

    if embed_type == "annonce" and game.msg_id:
        response = self.edit_embed(game.msg_id, embed, target)
    else:
        response = self.send_embed(embed, target)

    return response["id"]

GameService

Service layer for Game business logic.

Handles game creation, updates, status transitions, player registration, and Discord integration. Owns transaction boundaries (commits).

Source code in website/services/game.py

class GameService:
    """Service layer for Game business logic.

    Handles game creation, updates, status transitions, player registration,
    and Discord integration. Owns transaction boundaries (commits).
    """

    def __init__(
        self,
        repository=None,
        user_service=None,
        channel_service=None,
        session_service=None,
        trophy_service=None,
        discord_service=None,
    ):
        from website.services.discord import DiscordService

        self.repo = repository or GameRepository()
        self.user_service = user_service or UserService()
        self.channel_service = channel_service or ChannelService()
        self.session_service = session_service or GameSessionService()
        self.trophy_service = trophy_service or TrophyService()
        self.discord = discord_service or DiscordService()

    def get_by_id(self, game_id: int) -> Game:
        """Get game by ID.

        Args:
            game_id: Game ID.

        Returns:
            Game instance.

        Raises:
            NotFoundError: If game doesn't exist.
        """
        game = self.repo.get_by_id(game_id)
        if not game:
            raise NotFoundError(
                f"Game with id {game_id} not found",
                resource_type="Game",
                resource_id=game_id,
            )
        return game

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

        Args:
            slug: URL-safe game identifier.

        Returns:
            Game instance.

        Raises:
            NotFoundError: If game doesn't exist.
        """
        game = self.repo.get_by_slug(slug)
        if not game:
            raise NotFoundError(
                f"Game with slug '{slug}' not found",
                resource_type="Game",
                resource_id=slug,
            )
        return game

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

        Args:
            slug: URL-safe game identifier.

        Returns:
            Game instance.

        Raises:
            NotFound: Flask 404 error.
        """
        return self.repo.get_by_slug_or_404(slug)

    def generate_slug(self, name: str, gm_name: str) -> str:
        """Generate unique slug for a game.

        Args:
            name: Game name.
            gm_name: GM stable username (preferred) or display name (fallback).

        Returns:
            Unique URL-safe slug.
        """
        existing_slugs = self.repo.get_all_slugs()
        base_slug = slugify(f"{name}-par-{gm_name}")
        slug = base_slug
        i = 2
        while slug in existing_slugs:
            slug = f"{base_slug}-{i}"
            i += 1
        return slug

    def parse_game_type(self, type_value: str) -> tuple[str, Optional[int]]:
        """Parse game type value from form.

        Args:
            type_value: Type string, possibly "specialevent-<id>".

        Returns:
            Tuple of (game_type, special_event_id).
        """
        special_event_id = None
        game_type = type_value

        if type_value and type_value.startswith("specialevent-"):
            try:
                special_event_id = int(type_value.split("-", 1)[1])
            except (ValueError, IndexError):
                special_event_id = None
            game_type = "oneshot"  # all special events are treated as oneshots

        return game_type, special_event_id

    def create(
        self,
        data: dict,
        gm_id: str,
        status: str = "draft",
        create_resources: bool = False,
    ) -> Game:
        """Create a new game.

        Args:
            data: Game data dictionary from form.
            gm_id: GM user ID.
            status: Initial status (draft, open, closed).
            create_resources: Whether to create Discord resources (role, channel).

        Returns:
            Created Game instance.

        Raises:
            ValidationError: If data is invalid.
            DiscordAPIError: If Discord resource creation fails.
        """
        from config.constants import DEFAULT_TIMEFORMAT
        from website.utils.form_parsers import (
            get_ambience,
            get_classification,
            parse_restriction_tags,
        )

        try:
            # Parse special fields
            game_type, special_event_id = self.parse_game_type(data["type"])

            # Get GM name for slug
            gm = self.user_service.get_by_id(gm_id)

            # Create game instance
            game = Game(
                name=data["name"],
                type=game_type,
                special_event_id=special_event_id,
                length=data["length"],
                gm_id=gm_id,
                system_id=data["system"],
                vtt_id=data.get("vtt") or None,
                description=data["description"],
                restriction=data["restriction"],
                party_size=data["party_size"],
                xp=data["xp"],
                date=datetime.strptime(data["date"], DEFAULT_TIMEFORMAT),
                session_length=data["session_length"],
                frequency=data.get("frequency") or None,
                characters=data["characters"],
                classification=get_classification(),
                ambience=get_ambience(data),
                complement=data.get("complement"),
                status=status,
                img=data.get("img"),
                party_selection="party_selection" in data,
                restriction_tags=parse_restriction_tags(data),
            )

            # Generate unique slug
            game.slug = self.generate_slug(data["name"], gm.slug_name)

            # Add to session
            self.repo.add(game)
            db.session.flush()  # Ensure game.id is available

            logger.info(f"Game object created: {game.name} with slug {game.slug}")

            # Create Discord resources if requested
            if create_resources:
                self._setup_game_resources(game)
                logger.info("Game post-creation setup completed.")

            db.session.commit()
            log_game_event(
                "create",
                game.id,
                f"Annonce créée avec le statut {game.status}.",
                user_id=game.gm_id,
            )
            logger.info(f"Game saved in DB with ID: {game.id}")

            return game

        except ValidationError:
            db.session.rollback()
            raise
        except Exception as e:
            db.session.rollback()
            logger.error(f"Failed to create game: {e}", exc_info=True)
            # Rollback Discord resources if they were created
            if create_resources and hasattr(game, "role"):
                self._rollback_discord_resources(game)
            raise

    def _setup_game_resources(self, game: Game) -> None:
        """Set up Discord resources for a game (role, channel, session, channel message).

        Args:
            game: Game instance.

        Raises:
            DiscordAPIError: If Discord operations fail.
        """
        # Create initial game session
        self.session_service.create(
            game,
            game.date,
            game.date + timedelta(hours=float(game.session_length)),
        )
        logger.info("Initial game session created.")

        # Create Discord role
        game.role = self.discord.create_role(
            name="PJ_" + game.slug,
            permissions=PLAYER_ROLE_PERMISSION,
            color=Game.COLORS[game.type],
        )["id"]
        logger.info(f"Role created with ID: {game.role}")

        # Create Discord channel
        category = self.channel_service.get_category(game.type)
        game.channel = self.discord.create_channel(
            name=game.slug.lower(),
            parent_id=category.id,
            role_id=game.role,
            gm_id=game.gm_id,
        )["id"]
        logger.info(f"Channel created with ID: {game.channel} under category: {category.id}")

        self.channel_service.increment_size(category)

        # Post and pin initial message in the game channel
        msg_id = self.discord.send_game_embed(game, embed_type="annonce_details")
        self.discord.pin_message(msg_id, game.channel)
        logger.info("Initial channel message posted and pinned.")

    def _rollback_discord_resources(self, game: Game) -> None:
        """Rollback Discord resources on error.

        Args:
            game: Game instance with potentially created resources.
        """
        if game.channel:
            self.discord.delete_channel(game.channel)
            logger.info(f"Channel {game.channel} deleted")
        if game.role:
            self.discord.delete_role(game.role)
            logger.info(f"Role {game.role} deleted")

    def update(self, slug: str, data: dict, user_id: str | None = None) -> Game:
        """Update an existing game.

        Args:
            slug: Game slug.
            data: Updated game data.
            user_id: ID of the user performing the update.

        Returns:
            Updated Game instance.

        Raises:
            NotFoundError: If game doesn't exist.
            ValidationError: If data is invalid.
        """
        from config.constants import DEFAULT_TIMEFORMAT
        from website.utils.form_parsers import (
            get_ambience,
            get_classification,
            parse_restriction_tags,
        )

        game = self.get_by_slug(slug)

        try:
            # Only allow type/name changes if game is draft
            if game.status == "draft":
                game_type, special_event_id = self.parse_game_type(data["type"])
                game.type = game_type
                game.special_event_id = special_event_id
                game.name = data["name"]

            # Update fields
            game.system_id = data["system"]
            game.vtt_id = data.get("vtt") or None
            game.description = data["description"]
            game.date = datetime.strptime(data["date"], DEFAULT_TIMEFORMAT)
            game.length = data["length"]
            game.party_size = data["party_size"]
            game.party_selection = "party_selection" in data
            game.xp = data["xp"]
            game.session_length = data["session_length"]
            game.frequency = data.get("frequency") or None
            game.characters = data["characters"]
            game.classification = get_classification()
            game.ambience = get_ambience(data)
            game.complement = data.get("complement")
            game.img = data.get("img")
            game.restriction = data["restriction"]
            game.restriction_tags = parse_restriction_tags(data)

            db.session.commit()
            log_game_event(
                "edit", game.id, "Le contenu de l'annonce a été édité.", user_id=user_id
            )
            logger.info(f"Game {game.id} changes saved")

            # Update Discord embed if message exists
            if game.msg_id:
                try:
                    self.discord.send_game_embed(game, embed_type="annonce")
                    logger.info(f"Embed updated for game {game.id}")
                except DiscordAPIError as e:
                    logger.warning(f"Failed to update Discord embed for game {game.id}: {e}")

            return game

        except ValidationError:
            db.session.rollback()
            raise
        except Exception as e:
            db.session.rollback()
            logger.error(f"Failed to update game {game.id}: {e}", exc_info=True)
            raise

    def publish(self, slug: str, silent: bool = False, user_id: str | None = None) -> Game:
        """Publish a draft game to Discord.

        Args:
            slug: Game slug.
            silent: If True, don't send announcement (set to closed instead of open).
            user_id: ID of the user performing the publish.

        Returns:
            Published Game instance.

        Raises:
            NotFoundError: If game doesn't exist.
            ValidationError: If game is already published or is full.
            DiscordAPIError: If Discord operations fail.
        """
        game = self.get_by_slug(slug)

        if game.msg_id:
            raise ValidationError("Game is already published.", field="status")

        if len(game.players) >= game.party_size:
            raise ValidationError("Cannot publish a full game.", field="party_size")

        try:
            game.status = "closed" if silent else "open"

            # Set up resources if not already created
            if not game.role or not game.channel:
                self._setup_game_resources(game)

            # Send Discord announcement if not silent
            if not silent:
                game.msg_id = self.discord.send_game_embed(game, embed_type="annonce")
                logger.info(f"Discord embed sent with message ID: {game.msg_id}")

            db.session.commit()
            log_game_event(
                "edit",
                game.id,
                (
                    "L'annonce a été publiée et ouverte."
                    if not silent
                    else "L'annonce a été ouverte silencieusement."
                ),
                user_id=user_id,
            )
            logger.info(
                f"Game {game.id} published and {'opened' if not silent else 'opened silently'}."
            )

            return game

        except Exception as e:
            db.session.rollback()
            logger.error(f"Failed to publish game {game.id}: {e}", exc_info=True)
            # Rollback Discord resources if they were just created
            if not game.role or not game.channel:
                self._rollback_discord_resources(game)
            raise

    def close(self, slug: str, user_id: str | None = None) -> Game:
        """Close a game to new registrations.

        Args:
            slug: Game slug.
            user_id: ID of the user performing the close.

        Returns:
            Updated Game instance.

        Raises:
            NotFoundError: If game doesn't exist.
        """
        game = self.get_by_slug(slug)
        game.status = "closed"

        db.session.commit()
        log_game_event(
            "edit", game.id, "Le statut de l'annonce à changé en closed.", user_id=user_id
        )
        logger.info(f"Game status for {game.id} has been updated to closed")

        # Update Discord embed
        if game.msg_id:
            try:
                self.discord.send_game_embed(game, embed_type="annonce")
                logger.info(f"Embed updated due to status change for game {game.id}")
            except DiscordAPIError as e:
                logger.warning(f"Failed to update embed on status change for game {game.id}: {e}")

        return game

    def reopen(self, slug: str, user_id: str | None = None) -> Game:
        """Reopen a closed game.

        Args:
            slug: Game slug.
            user_id: ID of the user performing the reopen.

        Returns:
            Updated Game instance.

        Raises:
            NotFoundError: If game doesn't exist.
        """
        game = self.get_by_slug(slug)
        game.status = "open"

        db.session.commit()
        log_game_event(
            "edit", game.id, "Le statut de l'annonce à changé en open.", user_id=user_id
        )
        logger.info(f"Game status for {game.id} has been updated to open")

        # Update Discord embed
        if game.msg_id:
            try:
                self.discord.send_game_embed(game, embed_type="annonce")
                logger.info(f"Embed updated due to status change for game {game.id}")
            except DiscordAPIError as e:
                logger.warning(f"Failed to update embed on status change for game {game.id}: {e}")

        return game

    def archive(self, slug: str, award_trophies: bool = True, user_id: str | None = None) -> None:
        """Archive a game and clean up Discord resources.

        Args:
            slug: Game slug.
            award_trophies: Whether to award trophies to participants.
            user_id: ID of the user performing the archive.

        Raises:
            NotFoundError: If game doesn't exist.
        """
        game = self.get_by_slug(slug)
        game.status = "archived"

        db.session.commit()
        log_game_event(
            "edit", game.id, "Le statut de l'annonce à changé en archived.", user_id=user_id
        )
        logger.info(f"Game status for {game.id} has been updated to archived")

        # Award trophies
        msg = "Annonce archivée."
        if award_trophies:
            self._award_game_trophies(game)
            msg += " Badges distribués."
        else:
            msg += " Badges non-distribués."

        # Clean up Discord resources
        self._cleanup_discord_resources(game)
        self._delete_game_message(game)

        log_game_event("delete", game.id, msg, user_id=user_id)

    def _award_game_trophies(self, game: Game) -> None:
        """Award trophies to GM and players.

        Args:
            game: Game instance.
        """
        trophy_map = {
            "oneshot": (BADGE_OS_GM_ID, BADGE_OS_ID),
            "campaign": (BADGE_CAMPAIGN_GM_ID, BADGE_CAMPAIGN_ID),
        }
        gm_trophy, player_trophy = trophy_map.get(game.type, (None, None))
        if gm_trophy:
            try:
                self.trophy_service.award(user_id=game.gm.id, trophy_id=gm_trophy)
                for user in game.players:
                    self.trophy_service.award(user_id=user.id, trophy_id=player_trophy)
            except Exception as e:
                logger.error(f"Failed to award trophies for game {game.id}: {e}")

    def _cleanup_discord_resources(self, game: Game) -> None:
        """Clean up Discord resources for a game.

        Args:
            game: Game instance.
        """
        self.channel_service.adjust_category_size(self.discord, game)

        try:
            self.discord.delete_channel(game.channel)
            logger.info(f"Game {game.id} channel {game.channel} has been deleted")
        except DiscordAPIError as e:
            logger.warning(f"Failed to delete channel for game {game.id}: {e}")

        try:
            self.discord.delete_role(game.role)
            logger.info(f"Game {game.id} role {game.role} has been deleted")
        except DiscordAPIError as e:
            logger.warning(f"Failed to delete role for game {game.id}: {e}")

    def _delete_game_message(self, game: Game) -> None:
        """Delete Discord announcement message.

        Args:
            game: Game instance.
        """
        if not game.msg_id:
            return

        try:
            self.discord.delete_message(game.msg_id, current_app.config["POSTS_CHANNEL_ID"])
            game.msg_id = None
            db.session.commit()
            logger.info(f"Discord embed message deleted for archived game {game.id}")
        except DiscordAPIError as e:
            logger.warning(f"Failed to delete message for archived game {game.id}: {e}")

    def delete(self, slug: str) -> None:
        """Delete a game permanently.

        Args:
            slug: Game slug.

        Raises:
            NotFoundError: If game doesn't exist.
        """
        game = self.get_by_slug(slug)
        self.repo.delete_by_id(game.id)
        db.session.commit()
        logger.info(f"Game {game.id} has been deleted.")

    def register_player(self, slug: str, user_id: str, force: bool = False) -> Game:
        """Register a player to a game (concurrent-safe).

        Args:
            slug: Game slug.
            user_id: User ID to register.
            force: If True, bypass capacity and status checks.

        Returns:
            Updated Game instance.

        Raises:
            NotFoundError: If game doesn't exist.
            DuplicateRegistrationError: If user is already registered.
            GameFullError: If game is at capacity and force is False.
            GameClosedError: If game is closed and force is False.
        """
        game = self.get_by_slug(slug)
        user = self.user_service.get_by_id(user_id)

        try:
            # Lock the game row for update
            locked_game = self.repo.get_for_update(game.id)
            if not locked_game:
                raise NotFoundError(
                    f"Game with id {game.id} not found",
                    resource_type="Game",
                    resource_id=game.id,
                )

            # Check if already registered
            if user in locked_game.players:
                raise DuplicateRegistrationError(
                    "User is already registered for this game.",
                    game_id=locked_game.id,
                    user_id=user.id,
                )

            # Check capacity
            if len(locked_game.players) >= locked_game.party_size and not force:
                raise GameFullError(
                    "Game is full.",
                    game_id=locked_game.id,
                    max_players=locked_game.party_size,
                )

            # Check status
            if locked_game.status == "closed" and not force:
                raise GameClosedError(
                    "Game is closed for registration.",
                    game_id=locked_game.id,
                )

            # Add player
            locked_game.players.append(user)

            # Auto-close if full
            if (
                len(locked_game.players) >= locked_game.party_size
                and not locked_game.party_selection
            ):
                locked_game.status = "closed"
                if locked_game.msg_id:
                    try:
                        self.discord.send_game_embed(locked_game, embed_type="annonce")
                        logger.info(
                            f"Embed updated due to status change for game {locked_game.id}"
                        )
                    except DiscordAPIError as e:
                        logger.warning(
                            "Failed to update embed on status change "
                            f"for game {locked_game.id}: {e}"
                        )
                log_game_event(
                    "edit",
                    locked_game.id,
                    "Annonce fermée automatiquement après avoir atteint le nombre "
                    f"max de joueur·euses ({locked_game.party_size}).",
                )
                logger.info(f"Game status for {locked_game.id} has been updated to closed")

            db.session.commit()

            # Log event
            if force:
                log_game_event(
                    "register",
                    locked_game.id,
                    "Le·a joueur·euse a été ajouté·e à la partie par le MJ.",
                    user_id=user.id,
                )
            else:
                log_game_event(
                    "register",
                    locked_game.id,
                    "Le·a joueur·euse s'est inscrit·e sur l'annonce.",
                    user_id=user.id,
                )

            logger.info(f"User {user.id} registered to Game {locked_game.id}")

            # Add Discord role
            self.discord.add_role_to_user(user.id, locked_game.role)
            logger.info(f"Role {locked_game.role} added to user {user.id}")

            # Send registration embed
            self.discord.send_game_embed(locked_game, embed_type="register", player=user.id)

            return locked_game

        except (DuplicateRegistrationError, GameFullError, GameClosedError):
            db.session.rollback()
            raise
        except SQLAlchemyError:
            db.session.rollback()
            logger.exception("Failed to register user due to DB error.")
            raise

    def unregister_player(self, slug: str, user_id: str) -> Game:
        """Unregister a player from a game.

        Args:
            slug: Game slug.
            user_id: User ID to unregister.

        Returns:
            Updated Game instance.

        Raises:
            NotFoundError: If game or user doesn't exist.
            ValidationError: If user is not registered.
        """
        game = self.get_by_slug(slug)
        user = self.user_service.get_by_id(user_id)

        if user not in game.players:
            raise ValidationError(
                "User is not registered for this game.",
                field="user_id",
            )

        game.players.remove(user)

        # Reopen if it was full
        if (
            game.status == "closed"
            and len(game.players) < game.party_size
            and not game.party_selection
        ):
            game.status = "open"
            log_game_event(
                "edit",
                game.id,
                "Annonce rouverte automatiquement après désinscription.",
            )

        db.session.commit()
        logger.info(f"User {user.id} removed from Game {game.id}")

        # Remove Discord role
        self.discord.remove_role_from_user(user.id, game.role)
        logger.info(f"Role {game.role} removed from Player {user.id}")

        log_game_event(
            "unregister",
            game.id,
            "Le·a joueur·euse a été désinscrit·e de l'annonce.",
            user_id=user.id,
        )

        return game

    def clone(self, slug: str) -> dict:
        """Clone a game (return data dict for form prefill).

        Args:
            slug: Game slug to clone.

        Returns:
            Dict with game data for form.

        Raises:
            NotFoundError: If game doesn't exist.
        """
        game = self.get_by_slug(slug)
        return game.to_dict(include_relationships=False)

    def is_player(self, game: Game, user_id: str) -> bool:
        """Check if a user is registered as a player in a game.

        Args:
            game: Game instance.
            user_id: User ID to check.

        Returns:
            True if the user is a registered player.
        """
        return any(p.id == user_id for p in game.players)

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

        Args:
            filters: Search filters dict.
            page: Page number.
            per_page: Items per page.
            user_payload: User auth payload.

        Returns:
            Tuple of (games list, total count).
        """
        return self.repo.search(filters, page, per_page, user_payload)

get_by_id(game_id)

Get game by ID.

Parameters:

Name	Type	Description	Default
game_id	int	Game ID.	required

Returns:

Type	Description
Game	Game instance.

Raises:

Type	Description
NotFoundError	If game doesn't exist.

Source code in website/services/game.py

def get_by_id(self, game_id: int) -> Game:
    """Get game by ID.

    Args:
        game_id: Game ID.

    Returns:
        Game instance.

    Raises:
        NotFoundError: If game doesn't exist.
    """
    game = self.repo.get_by_id(game_id)
    if not game:
        raise NotFoundError(
            f"Game with id {game_id} not found",
            resource_type="Game",
            resource_id=game_id,
        )
    return game

get_by_slug(slug)

Get game by slug.

Parameters:

Name	Type	Description	Default
slug	str	URL-safe game identifier.	required

Returns:

Type	Description
Game	Game instance.

Raises:

Type	Description
NotFoundError	If game doesn't exist.

Source code in website/services/game.py

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

    Args:
        slug: URL-safe game identifier.

    Returns:
        Game instance.

    Raises:
        NotFoundError: If game doesn't exist.
    """
    game = self.repo.get_by_slug(slug)
    if not game:
        raise NotFoundError(
            f"Game with slug '{slug}' not found",
            resource_type="Game",
            resource_id=slug,
        )
    return game

get_by_slug_or_404(slug)

Get game by slug or raise 404 (for Flask routes).

Parameters:

Name	Type	Description	Default
slug	str	URL-safe game identifier.	required

Returns:

Type	Description
Game	Game instance.

Raises:

Type	Description
NotFound	Flask 404 error.

Source code in website/services/game.py

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

    Args:
        slug: URL-safe game identifier.

    Returns:
        Game instance.

    Raises:
        NotFound: Flask 404 error.
    """
    return self.repo.get_by_slug_or_404(slug)

generate_slug(name, gm_name)

Generate unique slug for a game.

Parameters:

Name	Type	Description	Default
name	str	Game name.	required
gm_name	str	GM stable username (preferred) or display name (fallback).	required

Returns:

Type	Description
str	Unique URL-safe slug.

Source code in website/services/game.py

def generate_slug(self, name: str, gm_name: str) -> str:
    """Generate unique slug for a game.

    Args:
        name: Game name.
        gm_name: GM stable username (preferred) or display name (fallback).

    Returns:
        Unique URL-safe slug.
    """
    existing_slugs = self.repo.get_all_slugs()
    base_slug = slugify(f"{name}-par-{gm_name}")
    slug = base_slug
    i = 2
    while slug in existing_slugs:
        slug = f"{base_slug}-{i}"
        i += 1
    return slug

parse_game_type(type_value)

Parse game type value from form.

Parameters:

Name	Type	Description	Default
type_value	str	Type string, possibly "specialevent-".	required

Returns:

Type	Description
tuple[str, Optional[int]]	Tuple of (game_type, special_event_id).

Source code in website/services/game.py

def parse_game_type(self, type_value: str) -> tuple[str, Optional[int]]:
    """Parse game type value from form.

    Args:
        type_value: Type string, possibly "specialevent-<id>".

    Returns:
        Tuple of (game_type, special_event_id).
    """
    special_event_id = None
    game_type = type_value

    if type_value and type_value.startswith("specialevent-"):
        try:
            special_event_id = int(type_value.split("-", 1)[1])
        except (ValueError, IndexError):
            special_event_id = None
        game_type = "oneshot"  # all special events are treated as oneshots

    return game_type, special_event_id

create(data, gm_id, status='draft', create_resources=False)

Create a new game.

Parameters:

Name	Type	Description	Default
data	dict	Game data dictionary from form.	required
gm_id	str	GM user ID.	required
status	str	Initial status (draft, open, closed).	'draft'
create_resources	bool	Whether to create Discord resources (role, channel).	False

Returns:

Type	Description
Game	Created Game instance.

Raises:

Type	Description
ValidationError	If data is invalid.
DiscordAPIError	If Discord resource creation fails.

Source code in website/services/game.py

def create(
    self,
    data: dict,
    gm_id: str,
    status: str = "draft",
    create_resources: bool = False,
) -> Game:
    """Create a new game.

    Args:
        data: Game data dictionary from form.
        gm_id: GM user ID.
        status: Initial status (draft, open, closed).
        create_resources: Whether to create Discord resources (role, channel).

    Returns:
        Created Game instance.

    Raises:
        ValidationError: If data is invalid.
        DiscordAPIError: If Discord resource creation fails.
    """
    from config.constants import DEFAULT_TIMEFORMAT
    from website.utils.form_parsers import (
        get_ambience,
        get_classification,
        parse_restriction_tags,
    )

    try:
        # Parse special fields
        game_type, special_event_id = self.parse_game_type(data["type"])

        # Get GM name for slug
        gm = self.user_service.get_by_id(gm_id)

        # Create game instance
        game = Game(
            name=data["name"],
            type=game_type,
            special_event_id=special_event_id,
            length=data["length"],
            gm_id=gm_id,
            system_id=data["system"],
            vtt_id=data.get("vtt") or None,
            description=data["description"],
            restriction=data["restriction"],
            party_size=data["party_size"],
            xp=data["xp"],
            date=datetime.strptime(data["date"], DEFAULT_TIMEFORMAT),
            session_length=data["session_length"],
            frequency=data.get("frequency") or None,
            characters=data["characters"],
            classification=get_classification(),
            ambience=get_ambience(data),
            complement=data.get("complement"),
            status=status,
            img=data.get("img"),
            party_selection="party_selection" in data,
            restriction_tags=parse_restriction_tags(data),
        )

        # Generate unique slug
        game.slug = self.generate_slug(data["name"], gm.slug_name)

        # Add to session
        self.repo.add(game)
        db.session.flush()  # Ensure game.id is available

        logger.info(f"Game object created: {game.name} with slug {game.slug}")

        # Create Discord resources if requested
        if create_resources:
            self._setup_game_resources(game)
            logger.info("Game post-creation setup completed.")

        db.session.commit()
        log_game_event(
            "create",
            game.id,
            f"Annonce créée avec le statut {game.status}.",
            user_id=game.gm_id,
        )
        logger.info(f"Game saved in DB with ID: {game.id}")

        return game

    except ValidationError:
        db.session.rollback()
        raise
    except Exception as e:
        db.session.rollback()
        logger.error(f"Failed to create game: {e}", exc_info=True)
        # Rollback Discord resources if they were created
        if create_resources and hasattr(game, "role"):
            self._rollback_discord_resources(game)
        raise

update(slug, data, user_id=None)

Update an existing game.

Parameters:

Name	Type	Description	Default
slug	str	Game slug.	required
data	dict	Updated game data.	required
user_id	str | None	ID of the user performing the update.	None

Returns:

Type	Description
Game	Updated Game instance.

Raises:

Type	Description
NotFoundError	If game doesn't exist.
ValidationError	If data is invalid.

Source code in website/services/game.py

def update(self, slug: str, data: dict, user_id: str | None = None) -> Game:
    """Update an existing game.

    Args:
        slug: Game slug.
        data: Updated game data.
        user_id: ID of the user performing the update.

    Returns:
        Updated Game instance.

    Raises:
        NotFoundError: If game doesn't exist.
        ValidationError: If data is invalid.
    """
    from config.constants import DEFAULT_TIMEFORMAT
    from website.utils.form_parsers import (
        get_ambience,
        get_classification,
        parse_restriction_tags,
    )

    game = self.get_by_slug(slug)

    try:
        # Only allow type/name changes if game is draft
        if game.status == "draft":
            game_type, special_event_id = self.parse_game_type(data["type"])
            game.type = game_type
            game.special_event_id = special_event_id
            game.name = data["name"]

        # Update fields
        game.system_id = data["system"]
        game.vtt_id = data.get("vtt") or None
        game.description = data["description"]
        game.date = datetime.strptime(data["date"], DEFAULT_TIMEFORMAT)
        game.length = data["length"]
        game.party_size = data["party_size"]
        game.party_selection = "party_selection" in data
        game.xp = data["xp"]
        game.session_length = data["session_length"]
        game.frequency = data.get("frequency") or None
        game.characters = data["characters"]
        game.classification = get_classification()
        game.ambience = get_ambience(data)
        game.complement = data.get("complement")
        game.img = data.get("img")
        game.restriction = data["restriction"]
        game.restriction_tags = parse_restriction_tags(data)

        db.session.commit()
        log_game_event(
            "edit", game.id, "Le contenu de l'annonce a été édité.", user_id=user_id
        )
        logger.info(f"Game {game.id} changes saved")

        # Update Discord embed if message exists
        if game.msg_id:
            try:
                self.discord.send_game_embed(game, embed_type="annonce")
                logger.info(f"Embed updated for game {game.id}")
            except DiscordAPIError as e:
                logger.warning(f"Failed to update Discord embed for game {game.id}: {e}")

        return game

    except ValidationError:
        db.session.rollback()
        raise
    except Exception as e:
        db.session.rollback()
        logger.error(f"Failed to update game {game.id}: {e}", exc_info=True)
        raise

publish(slug, silent=False, user_id=None)

Publish a draft game to Discord.

Parameters:

Name	Type	Description	Default
slug	str	Game slug.	required
silent	bool	If True, don't send announcement (set to closed instead of open).	False
user_id	str | None	ID of the user performing the publish.	None

Returns:

Type	Description
Game	Published Game instance.

Raises:

Type	Description
NotFoundError	If game doesn't exist.
ValidationError	If game is already published or is full.
DiscordAPIError	If Discord operations fail.

Source code in website/services/game.py

def publish(self, slug: str, silent: bool = False, user_id: str | None = None) -> Game:
    """Publish a draft game to Discord.

    Args:
        slug: Game slug.
        silent: If True, don't send announcement (set to closed instead of open).
        user_id: ID of the user performing the publish.

    Returns:
        Published Game instance.

    Raises:
        NotFoundError: If game doesn't exist.
        ValidationError: If game is already published or is full.
        DiscordAPIError: If Discord operations fail.
    """
    game = self.get_by_slug(slug)

    if game.msg_id:
        raise ValidationError("Game is already published.", field="status")

    if len(game.players) >= game.party_size:
        raise ValidationError("Cannot publish a full game.", field="party_size")

    try:
        game.status = "closed" if silent else "open"

        # Set up resources if not already created
        if not game.role or not game.channel:
            self._setup_game_resources(game)

        # Send Discord announcement if not silent
        if not silent:
            game.msg_id = self.discord.send_game_embed(game, embed_type="annonce")
            logger.info(f"Discord embed sent with message ID: {game.msg_id}")

        db.session.commit()
        log_game_event(
            "edit",
            game.id,
            (
                "L'annonce a été publiée et ouverte."
                if not silent
                else "L'annonce a été ouverte silencieusement."
            ),
            user_id=user_id,
        )
        logger.info(
            f"Game {game.id} published and {'opened' if not silent else 'opened silently'}."
        )

        return game

    except Exception as e:
        db.session.rollback()
        logger.error(f"Failed to publish game {game.id}: {e}", exc_info=True)
        # Rollback Discord resources if they were just created
        if not game.role or not game.channel:
            self._rollback_discord_resources(game)
        raise

close(slug, user_id=None)

Close a game to new registrations.

Parameters:

Name	Type	Description	Default
slug	str	Game slug.	required
user_id	str | None	ID of the user performing the close.	None

Returns:

Type	Description
Game	Updated Game instance.

Raises:

Type	Description
NotFoundError	If game doesn't exist.

Source code in website/services/game.py

def close(self, slug: str, user_id: str | None = None) -> Game:
    """Close a game to new registrations.

    Args:
        slug: Game slug.
        user_id: ID of the user performing the close.

    Returns:
        Updated Game instance.

    Raises:
        NotFoundError: If game doesn't exist.
    """
    game = self.get_by_slug(slug)
    game.status = "closed"

    db.session.commit()
    log_game_event(
        "edit", game.id, "Le statut de l'annonce à changé en closed.", user_id=user_id
    )
    logger.info(f"Game status for {game.id} has been updated to closed")

    # Update Discord embed
    if game.msg_id:
        try:
            self.discord.send_game_embed(game, embed_type="annonce")
            logger.info(f"Embed updated due to status change for game {game.id}")
        except DiscordAPIError as e:
            logger.warning(f"Failed to update embed on status change for game {game.id}: {e}")

    return game

reopen(slug, user_id=None)

Reopen a closed game.

Parameters:

Name	Type	Description	Default
slug	str	Game slug.	required
user_id	str | None	ID of the user performing the reopen.	None

Returns:

Type	Description
Game	Updated Game instance.

Raises:

Type	Description
NotFoundError	If game doesn't exist.

Source code in website/services/game.py

def reopen(self, slug: str, user_id: str | None = None) -> Game:
    """Reopen a closed game.

    Args:
        slug: Game slug.
        user_id: ID of the user performing the reopen.

    Returns:
        Updated Game instance.

    Raises:
        NotFoundError: If game doesn't exist.
    """
    game = self.get_by_slug(slug)
    game.status = "open"

    db.session.commit()
    log_game_event(
        "edit", game.id, "Le statut de l'annonce à changé en open.", user_id=user_id
    )
    logger.info(f"Game status for {game.id} has been updated to open")

    # Update Discord embed
    if game.msg_id:
        try:
            self.discord.send_game_embed(game, embed_type="annonce")
            logger.info(f"Embed updated due to status change for game {game.id}")
        except DiscordAPIError as e:
            logger.warning(f"Failed to update embed on status change for game {game.id}: {e}")

    return game

archive(slug, award_trophies=True, user_id=None)

Archive a game and clean up Discord resources.

Parameters:

Name	Type	Description	Default
slug	str	Game slug.	required
award_trophies	bool	Whether to award trophies to participants.	True
user_id	str | None	ID of the user performing the archive.	None

Raises:

Type	Description
NotFoundError	If game doesn't exist.

Source code in website/services/game.py

def archive(self, slug: str, award_trophies: bool = True, user_id: str | None = None) -> None:
    """Archive a game and clean up Discord resources.

    Args:
        slug: Game slug.
        award_trophies: Whether to award trophies to participants.
        user_id: ID of the user performing the archive.

    Raises:
        NotFoundError: If game doesn't exist.
    """
    game = self.get_by_slug(slug)
    game.status = "archived"

    db.session.commit()
    log_game_event(
        "edit", game.id, "Le statut de l'annonce à changé en archived.", user_id=user_id
    )
    logger.info(f"Game status for {game.id} has been updated to archived")

    # Award trophies
    msg = "Annonce archivée."
    if award_trophies:
        self._award_game_trophies(game)
        msg += " Badges distribués."
    else:
        msg += " Badges non-distribués."

    # Clean up Discord resources
    self._cleanup_discord_resources(game)
    self._delete_game_message(game)

    log_game_event("delete", game.id, msg, user_id=user_id)

delete(slug)

Delete a game permanently.

Parameters:

Name	Type	Description	Default
slug	str	Game slug.	required

Raises:

Type	Description
NotFoundError	If game doesn't exist.

Source code in website/services/game.py

def delete(self, slug: str) -> None:
    """Delete a game permanently.

    Args:
        slug: Game slug.

    Raises:
        NotFoundError: If game doesn't exist.
    """
    game = self.get_by_slug(slug)
    self.repo.delete_by_id(game.id)
    db.session.commit()
    logger.info(f"Game {game.id} has been deleted.")

register_player(slug, user_id, force=False)

Register a player to a game (concurrent-safe).

Parameters:

Name	Type	Description	Default
slug	str	Game slug.	required
user_id	str	User ID to register.	required
force	bool	If True, bypass capacity and status checks.	False

Returns:

Type	Description
Game	Updated Game instance.

Raises:

Type	Description
NotFoundError	If game doesn't exist.
DuplicateRegistrationError	If user is already registered.
GameFullError	If game is at capacity and force is False.
GameClosedError	If game is closed and force is False.

Source code in website/services/game.py

def register_player(self, slug: str, user_id: str, force: bool = False) -> Game:
    """Register a player to a game (concurrent-safe).

    Args:
        slug: Game slug.
        user_id: User ID to register.
        force: If True, bypass capacity and status checks.

    Returns:
        Updated Game instance.

    Raises:
        NotFoundError: If game doesn't exist.
        DuplicateRegistrationError: If user is already registered.
        GameFullError: If game is at capacity and force is False.
        GameClosedError: If game is closed and force is False.
    """
    game = self.get_by_slug(slug)
    user = self.user_service.get_by_id(user_id)

    try:
        # Lock the game row for update
        locked_game = self.repo.get_for_update(game.id)
        if not locked_game:
            raise NotFoundError(
                f"Game with id {game.id} not found",
                resource_type="Game",
                resource_id=game.id,
            )

        # Check if already registered
        if user in locked_game.players:
            raise DuplicateRegistrationError(
                "User is already registered for this game.",
                game_id=locked_game.id,
                user_id=user.id,
            )

        # Check capacity
        if len(locked_game.players) >= locked_game.party_size and not force:
            raise GameFullError(
                "Game is full.",
                game_id=locked_game.id,
                max_players=locked_game.party_size,
            )

        # Check status
        if locked_game.status == "closed" and not force:
            raise GameClosedError(
                "Game is closed for registration.",
                game_id=locked_game.id,
            )

        # Add player
        locked_game.players.append(user)

        # Auto-close if full
        if (
            len(locked_game.players) >= locked_game.party_size
            and not locked_game.party_selection
        ):
            locked_game.status = "closed"
            if locked_game.msg_id:
                try:
                    self.discord.send_game_embed(locked_game, embed_type="annonce")
                    logger.info(
                        f"Embed updated due to status change for game {locked_game.id}"
                    )
                except DiscordAPIError as e:
                    logger.warning(
                        "Failed to update embed on status change "
                        f"for game {locked_game.id}: {e}"
                    )
            log_game_event(
                "edit",
                locked_game.id,
                "Annonce fermée automatiquement après avoir atteint le nombre "
                f"max de joueur·euses ({locked_game.party_size}).",
            )
            logger.info(f"Game status for {locked_game.id} has been updated to closed")

        db.session.commit()

        # Log event
        if force:
            log_game_event(
                "register",
                locked_game.id,
                "Le·a joueur·euse a été ajouté·e à la partie par le MJ.",
                user_id=user.id,
            )
        else:
            log_game_event(
                "register",
                locked_game.id,
                "Le·a joueur·euse s'est inscrit·e sur l'annonce.",
                user_id=user.id,
            )

        logger.info(f"User {user.id} registered to Game {locked_game.id}")

        # Add Discord role
        self.discord.add_role_to_user(user.id, locked_game.role)
        logger.info(f"Role {locked_game.role} added to user {user.id}")

        # Send registration embed
        self.discord.send_game_embed(locked_game, embed_type="register", player=user.id)

        return locked_game

    except (DuplicateRegistrationError, GameFullError, GameClosedError):
        db.session.rollback()
        raise
    except SQLAlchemyError:
        db.session.rollback()
        logger.exception("Failed to register user due to DB error.")
        raise

unregister_player(slug, user_id)

Unregister a player from a game.

Parameters:

Name	Type	Description	Default
slug	str	Game slug.	required
user_id	str	User ID to unregister.	required

Returns:

Type	Description
Game	Updated Game instance.

Raises:

Type	Description
NotFoundError	If game or user doesn't exist.
ValidationError	If user is not registered.

Source code in website/services/game.py

def unregister_player(self, slug: str, user_id: str) -> Game:
    """Unregister a player from a game.

    Args:
        slug: Game slug.
        user_id: User ID to unregister.

    Returns:
        Updated Game instance.

    Raises:
        NotFoundError: If game or user doesn't exist.
        ValidationError: If user is not registered.
    """
    game = self.get_by_slug(slug)
    user = self.user_service.get_by_id(user_id)

    if user not in game.players:
        raise ValidationError(
            "User is not registered for this game.",
            field="user_id",
        )

    game.players.remove(user)

    # Reopen if it was full
    if (
        game.status == "closed"
        and len(game.players) < game.party_size
        and not game.party_selection
    ):
        game.status = "open"
        log_game_event(
            "edit",
            game.id,
            "Annonce rouverte automatiquement après désinscription.",
        )

    db.session.commit()
    logger.info(f"User {user.id} removed from Game {game.id}")

    # Remove Discord role
    self.discord.remove_role_from_user(user.id, game.role)
    logger.info(f"Role {game.role} removed from Player {user.id}")

    log_game_event(
        "unregister",
        game.id,
        "Le·a joueur·euse a été désinscrit·e de l'annonce.",
        user_id=user.id,
    )

    return game

clone(slug)

Clone a game (return data dict for form prefill).

Parameters:

Name	Type	Description	Default
slug	str	Game slug to clone.	required

Returns:

Type	Description
dict	Dict with game data for form.

Raises:

Type	Description
NotFoundError	If game doesn't exist.

Source code in website/services/game.py

def clone(self, slug: str) -> dict:
    """Clone a game (return data dict for form prefill).

    Args:
        slug: Game slug to clone.

    Returns:
        Dict with game data for form.

    Raises:
        NotFoundError: If game doesn't exist.
    """
    game = self.get_by_slug(slug)
    return game.to_dict(include_relationships=False)

is_player(game, user_id)

Check if a user is registered as a player in a game.

Parameters:

Name	Type	Description	Default
game	Game	Game instance.	required
user_id	str	User ID to check.	required

Returns:

Type	Description
bool	True if the user is a registered player.

Source code in website/services/game.py

def is_player(self, game: Game, user_id: str) -> bool:
    """Check if a user is registered as a player in a game.

    Args:
        game: Game instance.
        user_id: User ID to check.

    Returns:
        True if the user is a registered player.
    """
    return any(p.id == user_id for p in game.players)

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

Search games with filters.

Parameters:

Name	Type	Description	Default
filters	dict	Search filters dict.	required
page	int	Page number.	1
per_page	int	Items per page.	20
user_payload	Optional[dict]	User auth payload.	None

Returns:

Type	Description
tuple[list[Game], int]	Tuple of (games list, total count).

Source code in website/services/game.py

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

    Args:
        filters: Search filters dict.
        page: Page number.
        per_page: Items per page.
        user_payload: User auth payload.

    Returns:
        Tuple of (games list, total count).
    """
    return self.repo.search(filters, page, per_page, user_payload)

GameEventService

Service layer for game event logging.

Provides transaction-safe event recording for the game audit trail.

Source code in website/services/game_event.py

class GameEventService:
    """Service layer for game event logging.

    Provides transaction-safe event recording for the game audit trail.
    """

    def __init__(self, repository=None):
        self.repo = repository or GameEventRepository()

    def log_event(
        self, action: str, game_id: int, description: str | None = None, user_id: str | None = None
    ) -> GameEvent:
        """Log a game event and commit the transaction.

        Args:
            action: Event action type (create, edit, delete, etc.).
            game_id: ID of the related game.
            description: Optional human-readable description.
            user_id: Optional ID of the user that performed the action.

        Returns:
            Created GameEvent instance.
        """
        event = self.repo.log(action, game_id, description, user_id)
        db.session.commit()
        return event

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

Log a game event and commit the transaction.

Parameters:

Name	Type	Description	Default
action	str	Event action type (create, edit, delete, etc.).	required
game_id	int	ID of the related game.	required
description	str | None	Optional human-readable 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/services/game_event.py

def log_event(
    self, action: str, game_id: int, description: str | None = None, user_id: str | None = None
) -> GameEvent:
    """Log a game event and commit the transaction.

    Args:
        action: Event action type (create, edit, delete, etc.).
        game_id: ID of the related game.
        description: Optional human-readable description.
        user_id: Optional ID of the user that performed the action.

    Returns:
        Created GameEvent instance.
    """
    event = self.repo.log(action, game_id, description, user_id)
    db.session.commit()
    return event

GameSessionService

Service layer for GameSession operations.

Handles session creation, deletion, updates, and conflict detection.

Source code in website/services/game_session.py

class GameSessionService:
    """Service layer for GameSession operations.

    Handles session creation, deletion, updates, and conflict detection.
    """

    def __init__(self, repository=None):
        self.repo = repository or GameSessionRepository()

    def create(self, game: Game, start: datetime, end: datetime) -> GameSession:
        """Create a new game session.

        Args:
            game: Game instance to add the session to.
            start: Session start datetime.
            end: Session end datetime.

        Returns:
            Created GameSession instance.

        Raises:
            ValidationError: If start >= end.
            SessionConflictError: If the session overlaps with an existing one.
        """
        if start >= end:
            raise ValidationError("Session start must be before end time.")

        if self._has_conflict(game, start, end):
            raise SessionConflictError(
                "Session overlaps with an existing session.", game_id=game.id
            )

        session = GameSession(start=start, end=end)
        self.repo.add(session)
        game.sessions.append(session)
        db.session.commit()
        logger.info(f"Session added for game {game.id} from {start} to {end}")
        return session

    def delete(self, session: GameSession) -> None:
        """Delete a game session.

        Args:
            session: GameSession instance to delete.
        """
        game_id = session.game_id
        start = session.start
        end = session.end
        self.repo.delete(session)
        db.session.commit()
        logger.info(f"Session removed for game {game_id} from {start} to {end}")

    def update(self, session: GameSession, new_start: datetime, new_end: datetime) -> GameSession:
        """Update a session's start/end times.

        Args:
            session: Existing GameSession instance.
            new_start: New start datetime.
            new_end: New end datetime.

        Returns:
            Updated GameSession instance.

        Raises:
            ValidationError: If new_start >= new_end.
            SessionConflictError: If new times overlap another session.
        """
        if new_start >= new_end:
            raise ValidationError("Session start must be before end time.")

        game = session.game
        if self._has_conflict(game, new_start, new_end, exclude_session_id=session.id):
            raise SessionConflictError(
                "Session overlaps with an existing session.", game_id=game.id
            )

        session.start = new_start
        session.end = new_end
        db.session.commit()
        logger.info(f"Session {session.id} updated to {new_start} - {new_end}")
        return session

    def get_by_id_or_404(self, session_id: int) -> GameSession:
        """Get session by ID or abort with 404.

        Args:
            session_id: Session ID.

        Returns:
            GameSession instance.

        Raises:
            NotFound: Flask 404 error.
        """
        return self.repo.get_by_id_or_404(session_id)

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

        Args:
            start: Range start datetime.
            end: Range end datetime.

        Returns:
            List of GameSession instances within the range.
        """
        return self.repo.find_in_range(start, end)

    @cache.memoize(timeout=3600)
    def get_stats_for_period(self, year: int | None, month: int | None) -> dict:
        """Compute game statistics for a given month.

        Aggregates session data into per-system, per-game counts for
        oneshots and campaigns, along with GM participation.

        Args:
            year: Year to compute stats for, or None for current month.
            month: Month to compute stats for, or None for current month.

        Returns:
            Dict with keys: base_day, last_day, num_os, num_campaign,
            os_games, campaign_games, gm_names.
        """
        if year and month:
            base_day = datetime(year, month, 1)
        else:
            today = datetime.today()
            base_day = today.replace(day=1)

        last_day = datetime(
            base_day.year,
            base_day.month,
            calendar.monthrange(base_day.year, base_day.month)[1],
            23,
            59,
            59,
            999999,
        )

        sessions = self.find_in_range(base_day, last_day)

        num_os = 0
        num_campaign = 0
        os_games: dict = defaultdict(lambda: defaultdict(self._default_game_entry))
        campaign_games: dict = defaultdict(lambda: defaultdict(self._default_game_entry))
        gm_names: list[str] = []

        for session in sessions:
            game = session.game
            system = game.system.name
            slug = game.slug
            entry = {"name": game.name, "gm": game.gm.name, "count": 1}

            if game.type == "oneshot":
                num_os += 1
                if slug in os_games[system]:
                    os_games[system][slug]["count"] += 1
                else:
                    os_games[system][slug] = entry
            else:
                num_campaign += 1
                if slug in campaign_games[system]:
                    campaign_games[system][slug]["count"] += 1
                else:
                    campaign_games[system][slug] = entry

            gm_names.append(game.gm.name)

        return {
            "base_day": base_day,
            "last_day": last_day,
            "num_os": num_os,
            "num_campaign": num_campaign,
            "os_games": os_games,
            "campaign_games": campaign_games,
            "gm_names": gm_names,
        }

    @staticmethod
    def _default_game_entry():
        """Return a default game entry dict for stats aggregation."""
        return {"count": 0, "gm": ""}

    @staticmethod
    def _has_conflict(game, start_dt, end_dt, exclude_session_id=None):
        for s in game.sessions:
            if exclude_session_id and s.id == exclude_session_id:
                continue
            if not (end_dt <= s.start or start_dt >= s.end):
                return True
        return False

create(game, start, end)

Create a new game session.

Parameters:

Name	Type	Description	Default
game	Game	Game instance to add the session to.	required
start	datetime	Session start datetime.	required
end	datetime	Session end datetime.	required

Returns:

Type	Description
GameSession	Created GameSession instance.

Raises:

Type	Description
ValidationError	If start >= end.
SessionConflictError	If the session overlaps with an existing one.

Source code in website/services/game_session.py

def create(self, game: Game, start: datetime, end: datetime) -> GameSession:
    """Create a new game session.

    Args:
        game: Game instance to add the session to.
        start: Session start datetime.
        end: Session end datetime.

    Returns:
        Created GameSession instance.

    Raises:
        ValidationError: If start >= end.
        SessionConflictError: If the session overlaps with an existing one.
    """
    if start >= end:
        raise ValidationError("Session start must be before end time.")

    if self._has_conflict(game, start, end):
        raise SessionConflictError(
            "Session overlaps with an existing session.", game_id=game.id
        )

    session = GameSession(start=start, end=end)
    self.repo.add(session)
    game.sessions.append(session)
    db.session.commit()
    logger.info(f"Session added for game {game.id} from {start} to {end}")
    return session

delete(session)

Delete a game session.

Parameters:

Name	Type	Description	Default
session	GameSession	GameSession instance to delete.	required

Source code in website/services/game_session.py

def delete(self, session: GameSession) -> None:
    """Delete a game session.

    Args:
        session: GameSession instance to delete.
    """
    game_id = session.game_id
    start = session.start
    end = session.end
    self.repo.delete(session)
    db.session.commit()
    logger.info(f"Session removed for game {game_id} from {start} to {end}")

update(session, new_start, new_end)

Update a session's start/end times.

Parameters:

Name	Type	Description	Default
session	GameSession	Existing GameSession instance.	required
new_start	datetime	New start datetime.	required
new_end	datetime	New end datetime.	required

Returns:

Type	Description
GameSession	Updated GameSession instance.

Raises:

Type	Description
ValidationError	If new_start >= new_end.
SessionConflictError	If new times overlap another session.

Source code in website/services/game_session.py

def update(self, session: GameSession, new_start: datetime, new_end: datetime) -> GameSession:
    """Update a session's start/end times.

    Args:
        session: Existing GameSession instance.
        new_start: New start datetime.
        new_end: New end datetime.

    Returns:
        Updated GameSession instance.

    Raises:
        ValidationError: If new_start >= new_end.
        SessionConflictError: If new times overlap another session.
    """
    if new_start >= new_end:
        raise ValidationError("Session start must be before end time.")

    game = session.game
    if self._has_conflict(game, new_start, new_end, exclude_session_id=session.id):
        raise SessionConflictError(
            "Session overlaps with an existing session.", game_id=game.id
        )

    session.start = new_start
    session.end = new_end
    db.session.commit()
    logger.info(f"Session {session.id} updated to {new_start} - {new_end}")
    return session

get_by_id_or_404(session_id)

Get session by ID or abort with 404.

Parameters:

Name	Type	Description	Default
session_id	int	Session ID.	required

Returns:

Type	Description
GameSession	GameSession instance.

Raises:

Type	Description
NotFound	Flask 404 error.

Source code in website/services/game_session.py

def get_by_id_or_404(self, session_id: int) -> GameSession:
    """Get session by ID or abort with 404.

    Args:
        session_id: Session ID.

    Returns:
        GameSession instance.

    Raises:
        NotFound: Flask 404 error.
    """
    return self.repo.get_by_id_or_404(session_id)

find_in_range(start, end)

Find all sessions within a date range.

Parameters:

Name	Type	Description	Default
start	datetime	Range start datetime.	required
end	datetime	Range end datetime.	required

Returns:

Type	Description
list[GameSession]	List of GameSession instances within the range.

Source code in website/services/game_session.py

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

    Args:
        start: Range start datetime.
        end: Range end datetime.

    Returns:
        List of GameSession instances within the range.
    """
    return self.repo.find_in_range(start, end)

get_stats_for_period(year, month)

Compute game statistics for a given month.

Aggregates session data into per-system, per-game counts for oneshots and campaigns, along with GM participation.

Parameters:

Name	Type	Description	Default
year	int | None	Year to compute stats for, or None for current month.	required
month	int | None	Month to compute stats for, or None for current month.	required

Returns:

Type	Description
dict	Dict with keys: base_day, last_day, num_os, num_campaign,
dict	os_games, campaign_games, gm_names.

Source code in website/services/game_session.py

@cache.memoize(timeout=3600)
def get_stats_for_period(self, year: int | None, month: int | None) -> dict:
    """Compute game statistics for a given month.

    Aggregates session data into per-system, per-game counts for
    oneshots and campaigns, along with GM participation.

    Args:
        year: Year to compute stats for, or None for current month.
        month: Month to compute stats for, or None for current month.

    Returns:
        Dict with keys: base_day, last_day, num_os, num_campaign,
        os_games, campaign_games, gm_names.
    """
    if year and month:
        base_day = datetime(year, month, 1)
    else:
        today = datetime.today()
        base_day = today.replace(day=1)

    last_day = datetime(
        base_day.year,
        base_day.month,
        calendar.monthrange(base_day.year, base_day.month)[1],
        23,
        59,
        59,
        999999,
    )

    sessions = self.find_in_range(base_day, last_day)

    num_os = 0
    num_campaign = 0
    os_games: dict = defaultdict(lambda: defaultdict(self._default_game_entry))
    campaign_games: dict = defaultdict(lambda: defaultdict(self._default_game_entry))
    gm_names: list[str] = []

    for session in sessions:
        game = session.game
        system = game.system.name
        slug = game.slug
        entry = {"name": game.name, "gm": game.gm.name, "count": 1}

        if game.type == "oneshot":
            num_os += 1
            if slug in os_games[system]:
                os_games[system][slug]["count"] += 1
            else:
                os_games[system][slug] = entry
        else:
            num_campaign += 1
            if slug in campaign_games[system]:
                campaign_games[system][slug]["count"] += 1
            else:
                campaign_games[system][slug] = entry

        gm_names.append(game.gm.name)

    return {
        "base_day": base_day,
        "last_day": last_day,
        "num_os": num_os,
        "num_campaign": num_campaign,
        "os_games": os_games,
        "campaign_games": campaign_games,
        "gm_names": gm_names,
    }

SpecialEventService

Service layer for SpecialEvent business logic.

Handles creation, updates, deletion, and retrieval of special events. Manages transaction boundaries and validation.

Source code in website/services/special_event.py

class SpecialEventService:
    """Service layer for SpecialEvent business logic.

    Handles creation, updates, deletion, and retrieval of special events.
    Manages transaction boundaries and validation.
    """

    def __init__(self, repository=None):
        self.repo = repository or SpecialEventRepository()

    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.
        """
        return self.repo.get_all(active_only=active_only)

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

        Convenience method for dropdowns and context processors.

        Returns:
            List of active SpecialEvent instances ordered by name.
        """
        return self.repo.get_active()

    def get_by_id(self, id: int) -> SpecialEvent:
        """Get special event by ID.

        Args:
            id: Special event ID.

        Returns:
            SpecialEvent instance.

        Raises:
            NotFoundError: If special event with given ID doesn't exist.
        """
        event = self.repo.get_by_id(id)
        if not event:
            raise NotFoundError(
                f"SpecialEvent with id {id} not found",
                resource_type="SpecialEvent",
                resource_id=id,
            )
        return event

    def create(
        self, name: str, emoji: str = None, color: int = None, active: bool = False
    ) -> SpecialEvent:
        """Create a new special event.

        Args:
            name: Name of the special event (must be unique).
            emoji: Optional emoji for the event.
            color: Optional color as integer (e.g., 0xFF6600).
            active: Whether the event is active. Defaults to False.

        Returns:
            Created SpecialEvent instance.

        Raises:
            ValidationError: If name already exists or validation fails.
        """
        if self.repo.get_by_name(name):
            raise ValidationError("Special event name already exists.", field="name")

        event = SpecialEvent(name=name, emoji=emoji, color=color, active=active)
        self.repo.add(event)
        db.session.commit()
        return event

    def update(self, id: int, data: dict) -> SpecialEvent:
        """Update special event.

        Args:
            id: Special event ID.
            data: Dictionary of fields to update.

        Returns:
            Updated SpecialEvent instance.

        Raises:
            NotFoundError: If special event doesn't exist.
            ValidationError: If name conflicts with existing event.
        """
        event = self.repo.get_by_id_or_404(id)

        # Check for name uniqueness if name is being changed
        if "name" in data and data["name"] != event.name:
            existing = self.repo.get_by_name(data["name"])
            if existing:
                raise ValidationError("Special event name already exists.", field="name")

        event.update_from_dict(data)
        db.session.commit()
        return event

    def delete(self, id: int) -> None:
        """Delete special event.

        Args:
            id: Special event ID.

        Raises:
            NotFoundError: If special event doesn't exist.
        """
        event = self.repo.get_by_id_or_404(id)
        self.repo.delete(event)
        db.session.commit()

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/services/special_event.py

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.
    """
    return self.repo.get_all(active_only=active_only)

get_active()

Get all active special events.

Convenience method for dropdowns and context processors.

Returns:

Type	Description
list[SpecialEvent]	List of active SpecialEvent instances ordered by name.

Source code in website/services/special_event.py

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

    Convenience method for dropdowns and context processors.

    Returns:
        List of active SpecialEvent instances ordered by name.
    """
    return self.repo.get_active()

get_by_id(id)

Get special event by ID.

Parameters:

Name	Type	Description	Default
id	int	Special event ID.	required

Returns:

Type	Description
SpecialEvent	SpecialEvent instance.

Raises:

Type	Description
NotFoundError	If special event with given ID doesn't exist.

Source code in website/services/special_event.py

def get_by_id(self, id: int) -> SpecialEvent:
    """Get special event by ID.

    Args:
        id: Special event ID.

    Returns:
        SpecialEvent instance.

    Raises:
        NotFoundError: If special event with given ID doesn't exist.
    """
    event = self.repo.get_by_id(id)
    if not event:
        raise NotFoundError(
            f"SpecialEvent with id {id} not found",
            resource_type="SpecialEvent",
            resource_id=id,
        )
    return event

create(name, emoji=None, color=None, active=False)

Create a new special event.

Parameters:

Name	Type	Description	Default
name	str	Name of the special event (must be unique).	required
emoji	str	Optional emoji for the event.	None
color	int	Optional color as integer (e.g., 0xFF6600).	None
active	bool	Whether the event is active. Defaults to False.	False

Returns:

Type	Description
SpecialEvent	Created SpecialEvent instance.

Raises:

Type	Description
ValidationError	If name already exists or validation fails.

Source code in website/services/special_event.py

def create(
    self, name: str, emoji: str = None, color: int = None, active: bool = False
) -> SpecialEvent:
    """Create a new special event.

    Args:
        name: Name of the special event (must be unique).
        emoji: Optional emoji for the event.
        color: Optional color as integer (e.g., 0xFF6600).
        active: Whether the event is active. Defaults to False.

    Returns:
        Created SpecialEvent instance.

    Raises:
        ValidationError: If name already exists or validation fails.
    """
    if self.repo.get_by_name(name):
        raise ValidationError("Special event name already exists.", field="name")

    event = SpecialEvent(name=name, emoji=emoji, color=color, active=active)
    self.repo.add(event)
    db.session.commit()
    return event

update(id, data)

Update special event.

Parameters:

Name	Type	Description	Default
id	int	Special event ID.	required
data	dict	Dictionary of fields to update.	required

Returns:

Type	Description
SpecialEvent	Updated SpecialEvent instance.

Raises:

Type	Description
NotFoundError	If special event doesn't exist.
ValidationError	If name conflicts with existing event.

Source code in website/services/special_event.py

def update(self, id: int, data: dict) -> SpecialEvent:
    """Update special event.

    Args:
        id: Special event ID.
        data: Dictionary of fields to update.

    Returns:
        Updated SpecialEvent instance.

    Raises:
        NotFoundError: If special event doesn't exist.
        ValidationError: If name conflicts with existing event.
    """
    event = self.repo.get_by_id_or_404(id)

    # Check for name uniqueness if name is being changed
    if "name" in data and data["name"] != event.name:
        existing = self.repo.get_by_name(data["name"])
        if existing:
            raise ValidationError("Special event name already exists.", field="name")

    event.update_from_dict(data)
    db.session.commit()
    return event

delete(id)

Delete special event.

Parameters:

Name	Type	Description	Default
id	int	Special event ID.	required

Raises:

Type	Description
NotFoundError	If special event doesn't exist.

Source code in website/services/special_event.py

def delete(self, id: int) -> None:
    """Delete special event.

    Args:
        id: Special event ID.

    Raises:
        NotFoundError: If special event doesn't exist.
    """
    event = self.repo.get_by_id_or_404(id)
    self.repo.delete(event)
    db.session.commit()

SystemService

Service layer for System (RPG game system) operations.

Handles CRUD operations with cache invalidation.

Source code in website/services/system.py

class SystemService:
    """Service layer for System (RPG game system) operations.

    Handles CRUD operations with cache invalidation.
    """

    def __init__(self, repository=None):
        self.repo = repository or SystemRepository()

    @cache.memoize()
    def get_all(self) -> list[System]:
        """Get all systems ordered by name.

        Returns:
            List of System instances.
        """
        return self.repo.get_all_ordered()

    def get_by_id(self, id: int) -> System:
        """Get system by ID.

        Args:
            id: System ID.

        Returns:
            System instance.

        Raises:
            NotFoundError: If system does not exist.
        """
        system = self.repo.get_by_id(id)
        if not system:
            raise NotFoundError(
                f"System with id {id} not found",
                resource_type="System",
                resource_id=id,
            )
        return system

    def create(self, name: str, icon: str = None) -> System:
        """Create a new game system.

        Args:
            name: System name (must be unique).
            icon: Optional icon path.

        Returns:
            Created System instance.

        Raises:
            ValidationError: If name already exists.
        """
        if self.repo.get_by_name(name):
            raise ValidationError("System name already exists.", field="name")
        system = System(name=name, icon=icon)
        self.repo.add(system)
        db.session.commit()
        cache.delete_memoized(self.get_all)
        return system

    def update(self, id: int, data: dict) -> System:
        """Update an existing system.

        Args:
            id: System ID.
            data: Dictionary of fields to update.

        Returns:
            Updated System instance.
        """
        system = self.repo.get_by_id_or_404(id)
        system.update_from_dict(data)
        db.session.commit()
        cache.delete_memoized(self.get_all)
        return system

    def delete(self, id: int) -> None:
        """Delete a system.

        Args:
            id: System ID.
        """
        system = self.repo.get_by_id_or_404(id)
        self.repo.delete(system)
        db.session.commit()
        cache.delete_memoized(self.get_all)

get_all()

Get all systems ordered by name.

Returns:

Type	Description
list[System]	List of System instances.

Source code in website/services/system.py

@cache.memoize()
def get_all(self) -> list[System]:
    """Get all systems ordered by name.

    Returns:
        List of System instances.
    """
    return self.repo.get_all_ordered()

get_by_id(id)

Get system by ID.

Parameters:

Name	Type	Description	Default
id	int	System ID.	required

Returns:

Type	Description
System	System instance.

Raises:

Type	Description
NotFoundError	If system does not exist.

Source code in website/services/system.py

def get_by_id(self, id: int) -> System:
    """Get system by ID.

    Args:
        id: System ID.

    Returns:
        System instance.

    Raises:
        NotFoundError: If system does not exist.
    """
    system = self.repo.get_by_id(id)
    if not system:
        raise NotFoundError(
            f"System with id {id} not found",
            resource_type="System",
            resource_id=id,
        )
    return system

create(name, icon=None)

Create a new game system.

Parameters:

Name	Type	Description	Default
name	str	System name (must be unique).	required
icon	str	Optional icon path.	None

Returns:

Type	Description
System	Created System instance.

Raises:

Type	Description
ValidationError	If name already exists.

Source code in website/services/system.py

def create(self, name: str, icon: str = None) -> System:
    """Create a new game system.

    Args:
        name: System name (must be unique).
        icon: Optional icon path.

    Returns:
        Created System instance.

    Raises:
        ValidationError: If name already exists.
    """
    if self.repo.get_by_name(name):
        raise ValidationError("System name already exists.", field="name")
    system = System(name=name, icon=icon)
    self.repo.add(system)
    db.session.commit()
    cache.delete_memoized(self.get_all)
    return system

update(id, data)

Update an existing system.

Parameters:

Name	Type	Description	Default
id	int	System ID.	required
data	dict	Dictionary of fields to update.	required

Returns:

Type	Description
System	Updated System instance.

Source code in website/services/system.py

def update(self, id: int, data: dict) -> System:
    """Update an existing system.

    Args:
        id: System ID.
        data: Dictionary of fields to update.

    Returns:
        Updated System instance.
    """
    system = self.repo.get_by_id_or_404(id)
    system.update_from_dict(data)
    db.session.commit()
    cache.delete_memoized(self.get_all)
    return system

delete(id)

Delete a system.

Parameters:

Name	Type	Description	Default
id	int	System ID.	required

Source code in website/services/system.py

def delete(self, id: int) -> None:
    """Delete a system.

    Args:
        id: System ID.
    """
    system = self.repo.get_by_id_or_404(id)
    self.repo.delete(system)
    db.session.commit()
    cache.delete_memoized(self.get_all)

TrophyService

Service layer for Trophy business logic.

Handles awarding trophies, leaderboards, and user badge retrieval. Manages transaction boundaries and trophy-specific business rules.

Source code in website/services/trophy.py

class TrophyService:
    """Service layer for Trophy business logic.

    Handles awarding trophies, leaderboards, and user badge retrieval.
    Manages transaction boundaries and trophy-specific business rules.
    """

    def __init__(self, repository=None):
        self.repo = repository or TrophyRepository()

    def get_by_id(self, trophy_id: int) -> Trophy:
        """Get trophy by ID.

        Args:
            trophy_id: Trophy ID.

        Returns:
            Trophy instance.

        Raises:
            NotFoundError: If trophy with given ID doesn't exist.
        """
        trophy = self.repo.get_by_id(trophy_id)
        if not trophy:
            raise NotFoundError(
                f"Trophy with id {trophy_id} not found",
                resource_type="Trophy",
                resource_id=trophy_id,
            )
        return trophy

    def get_all(self) -> list[Trophy]:
        """Get all trophy definitions.

        Returns:
            List of all Trophy instances.
        """
        return self.repo.get_all()

    def award(self, user_id: str, trophy_id: int, amount: int = 1) -> UserTrophy:
        """Award a trophy to a user.

        Handles both unique and non-unique trophies according to business rules:
        - Unique trophies: Only awarded once (quantity = 1), additional awards are ignored
        - Non-unique trophies: Quantity is incremented by amount

        Args:
            user_id: User ID to award trophy to.
            trophy_id: Trophy ID to award.
            amount: Quantity to award (only applies to non-unique trophies). Defaults to 1.

        Returns:
            UserTrophy instance (created or updated).

        Raises:
            NotFoundError: If trophy doesn't exist.
        """
        trophy = self.get_by_id(trophy_id)
        user_trophy = self.repo.get_user_trophy(user_id, trophy_id)

        if trophy.unique:
            # Unique trophies: only award once
            if user_trophy is None:
                user_trophy = self.repo.award_trophy(user_id, trophy_id, amount=1)
                logger.info(f"User {user_id} got a trophy: {trophy.name}")
            else:
                logger.debug(f"User {user_id} already has unique trophy {trophy.name}, skipping")
        else:
            # Non-unique trophies: increment quantity
            user_trophy = self.repo.award_trophy(user_id, trophy_id, amount)
            logger.info(f"User {user_id} got a trophy: {trophy.name} (x{amount})")

        db.session.commit()
        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.

        Raises:
            NotFoundError: If trophy doesn't exist.
        """
        # Verify trophy exists
        self.get_by_id(trophy_id)
        return self.repo.get_leaderboard(trophy_id, limit)

    def get_user_badges(self, user_id: str) -> list[dict]:
        """Get all trophies/badges for a user.

        Args:
            user_id: User ID.

        Returns:
            List of dicts with keys: name, icon, quantity.
        """
        user_trophies = (
            self.repo.session.query(UserTrophy).filter_by(user_id=user_id).join(Trophy).all()
        )

        return [
            {
                "name": ut.trophy.name,
                "icon": ut.trophy.icon,
                "quantity": ut.quantity,
            }
            for ut in user_trophies
        ]

get_by_id(trophy_id)

Get trophy by ID.

Parameters:

Name	Type	Description	Default
trophy_id	int	Trophy ID.	required

Returns:

Type	Description
Trophy	Trophy instance.

Raises:

Type	Description
NotFoundError	If trophy with given ID doesn't exist.

Source code in website/services/trophy.py

def get_by_id(self, trophy_id: int) -> Trophy:
    """Get trophy by ID.

    Args:
        trophy_id: Trophy ID.

    Returns:
        Trophy instance.

    Raises:
        NotFoundError: If trophy with given ID doesn't exist.
    """
    trophy = self.repo.get_by_id(trophy_id)
    if not trophy:
        raise NotFoundError(
            f"Trophy with id {trophy_id} not found",
            resource_type="Trophy",
            resource_id=trophy_id,
        )
    return trophy

get_all()

Get all trophy definitions.

Returns:

Type	Description
list[Trophy]	List of all Trophy instances.

Source code in website/services/trophy.py

def get_all(self) -> list[Trophy]:
    """Get all trophy definitions.

    Returns:
        List of all Trophy instances.
    """
    return self.repo.get_all()

award(user_id, trophy_id, amount=1)

Award a trophy to a user.

Handles both unique and non-unique trophies according to business rules: - Unique trophies: Only awarded once (quantity = 1), additional awards are ignored - Non-unique trophies: Quantity is incremented by amount

Parameters:

Name	Type	Description	Default
user_id	str	User ID to award trophy to.	required
trophy_id	int	Trophy ID to award.	required
amount	int	Quantity to award (only applies to non-unique trophies). Defaults to 1.	1

Returns:

Type	Description
UserTrophy	UserTrophy instance (created or updated).

Raises:

Type	Description
NotFoundError	If trophy doesn't exist.

Source code in website/services/trophy.py

def award(self, user_id: str, trophy_id: int, amount: int = 1) -> UserTrophy:
    """Award a trophy to a user.

    Handles both unique and non-unique trophies according to business rules:
    - Unique trophies: Only awarded once (quantity = 1), additional awards are ignored
    - Non-unique trophies: Quantity is incremented by amount

    Args:
        user_id: User ID to award trophy to.
        trophy_id: Trophy ID to award.
        amount: Quantity to award (only applies to non-unique trophies). Defaults to 1.

    Returns:
        UserTrophy instance (created or updated).

    Raises:
        NotFoundError: If trophy doesn't exist.
    """
    trophy = self.get_by_id(trophy_id)
    user_trophy = self.repo.get_user_trophy(user_id, trophy_id)

    if trophy.unique:
        # Unique trophies: only award once
        if user_trophy is None:
            user_trophy = self.repo.award_trophy(user_id, trophy_id, amount=1)
            logger.info(f"User {user_id} got a trophy: {trophy.name}")
        else:
            logger.debug(f"User {user_id} already has unique trophy {trophy.name}, skipping")
    else:
        # Non-unique trophies: increment quantity
        user_trophy = self.repo.award_trophy(user_id, trophy_id, amount)
        logger.info(f"User {user_id} got a trophy: {trophy.name} (x{amount})")

    db.session.commit()
    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.

Raises:

Type	Description
NotFoundError	If trophy doesn't exist.

Source code in website/services/trophy.py

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.

    Raises:
        NotFoundError: If trophy doesn't exist.
    """
    # Verify trophy exists
    self.get_by_id(trophy_id)
    return self.repo.get_leaderboard(trophy_id, limit)

get_user_badges(user_id)

Get all trophies/badges for a user.

Parameters:

Name	Type	Description	Default
user_id	str	User ID.	required

Returns:

Type	Description
list[dict]	List of dicts with keys: name, icon, quantity.

Source code in website/services/trophy.py

def get_user_badges(self, user_id: str) -> list[dict]:
    """Get all trophies/badges for a user.

    Args:
        user_id: User ID.

    Returns:
        List of dicts with keys: name, icon, quantity.
    """
    user_trophies = (
        self.repo.session.query(UserTrophy).filter_by(user_id=user_id).join(Trophy).all()
    )

    return [
        {
            "name": ut.trophy.name,
            "icon": ut.trophy.icon,
            "quantity": ut.quantity,
        }
        for ut in user_trophies
    ]

UserService

Service layer for User operations.

Handles user retrieval, creation, and Discord profile management.

Source code in website/services/user.py

class UserService:
    """Service layer for User operations.

    Handles user retrieval, creation, and Discord profile management.
    """

    def __init__(self, repository=None):
        self.repo = repository or UserRepository()

    def get_by_id(self, user_id: str) -> User:
        """Get user by ID.

        Args:
            user_id: Discord user ID.

        Returns:
            User instance.

        Raises:
            NotFoundError: If user does not exist.
        """
        user = self.repo.get_by_id(user_id)
        if not user:
            raise NotFoundError(
                f"User with id {user_id} not found",
                resource_type="User",
                resource_id=user_id,
            )
        return user

    def get_or_create(
        self, user_id: str, name: str = "Inconnu", username: str | None = None
    ) -> tuple[User, bool]:
        """Get an existing user or create a new one.

        Args:
            user_id: Discord user ID.
            name: Display name for new users. Defaults to 'Inconnu'.
            username: Stable Discord username (optional).

        Returns:
            Tuple of (User, created) where created is True if the user was new.
        """
        user = self.repo.get_by_id(user_id)
        if user:
            return user, False
        user = User(id=user_id, name=name, username=username)
        self.repo.add(user)
        db.session.commit()
        user.init_on_load()
        return user, True

    def get_all(self) -> list[User]:
        """Get all users.

        Returns:
            List of all User instances.
        """
        return self.repo.get_all()

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

        Returns:
            List of active User instances.
        """
        return self.repo.get_active_users()

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

        Lightweight query that avoids loading full ORM objects.

        Returns:
            List of active user ID strings.
        """
        return self.repo.get_active_user_ids()

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

        Returns:
            List of inactive User instances.
        """
        return self.repo.get_inactive_users()

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

        Lightweight query that avoids loading full ORM objects.

        Returns:
            List of inactive user ID strings.
        """
        return self.repo.get_inactive_user_ids()

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

        Args:
            ids: List of user ID strings.

        Returns:
            List of User instances matching the given IDs.
        """
        return self.repo.get_by_ids(ids)

    @staticmethod
    def get_user_profile(user_id: str, force_refresh: bool = False) -> dict:
        """Fetch a user's Discord profile.

        Args:
            user_id: Discord user ID.
            force_refresh: If True, bypass cache and fetch from Discord.

        Returns:
            Dict with 'name', 'avatar', and optionally 'not_found' keys.
        """
        return _get_user_profile(user_id, force_refresh=force_refresh)

    def mark_inactive(self, user_id: str) -> User:
        """Mark a user as inactive (left the Discord server).

        Args:
            user_id: Discord user ID.

        Returns:
            Updated User instance.

        Raises:
            NotFoundError: If user does not exist.
        """
        user = self.get_by_id(user_id)
        user.not_player_as_of = datetime.now(timezone.utc)
        db.session.commit()
        return user

    def clear_inactive(self, user_id: str) -> User:
        """Clear the inactive flag for a user (they have rejoined).

        Args:
            user_id: Discord user ID.

        Returns:
            Updated User instance.

        Raises:
            NotFoundError: If user does not exist.
        """
        user = self.get_by_id(user_id)
        user.not_player_as_of = None
        db.session.commit()
        return user

get_by_id(user_id)

Get user by ID.

Parameters:

Name	Type	Description	Default
user_id	str	Discord user ID.	required

Returns:

Type	Description
User	User instance.

Raises:

Type	Description
NotFoundError	If user does not exist.

Source code in website/services/user.py

def get_by_id(self, user_id: str) -> User:
    """Get user by ID.

    Args:
        user_id: Discord user ID.

    Returns:
        User instance.

    Raises:
        NotFoundError: If user does not exist.
    """
    user = self.repo.get_by_id(user_id)
    if not user:
        raise NotFoundError(
            f"User with id {user_id} not found",
            resource_type="User",
            resource_id=user_id,
        )
    return user

get_or_create(user_id, name='Inconnu', username=None)

Get an existing user or create a new one.

Parameters:

Name	Type	Description	Default
user_id	str	Discord user ID.	required
name	str	Display name for new users. Defaults to 'Inconnu'.	'Inconnu'
username	str | None	Stable Discord username (optional).	None

Returns:

Type	Description
tuple[User, bool]	Tuple of (User, created) where created is True if the user was new.

Source code in website/services/user.py

def get_or_create(
    self, user_id: str, name: str = "Inconnu", username: str | None = None
) -> tuple[User, bool]:
    """Get an existing user or create a new one.

    Args:
        user_id: Discord user ID.
        name: Display name for new users. Defaults to 'Inconnu'.
        username: Stable Discord username (optional).

    Returns:
        Tuple of (User, created) where created is True if the user was new.
    """
    user = self.repo.get_by_id(user_id)
    if user:
        return user, False
    user = User(id=user_id, name=name, username=username)
    self.repo.add(user)
    db.session.commit()
    user.init_on_load()
    return user, True

get_all()

Get all users.

Returns:

Type	Description
list[User]	List of all User instances.

Source code in website/services/user.py

def get_all(self) -> list[User]:
    """Get all users.

    Returns:
        List of all User instances.
    """
    return self.repo.get_all()

get_active_users()

Get all users not marked as inactive.

Returns:

Type	Description
list[User]	List of active User instances.

Source code in website/services/user.py

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

    Returns:
        List of active User instances.
    """
    return self.repo.get_active_users()

get_active_user_ids()

Get IDs of all users not marked as inactive.

Lightweight query that avoids loading full ORM objects.

Returns:

Type	Description
list[str]	List of active user ID strings.

Source code in website/services/user.py

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

    Lightweight query that avoids loading full ORM objects.

    Returns:
        List of active user ID strings.
    """
    return self.repo.get_active_user_ids()

get_inactive_users()

Get all users marked as inactive.

Returns:

Type	Description
list[User]	List of inactive User instances.

Source code in website/services/user.py

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

    Returns:
        List of inactive User instances.
    """
    return self.repo.get_inactive_users()

get_inactive_user_ids()

Get IDs of all users marked as inactive.

Lightweight query that avoids loading full ORM objects.

Returns:

Type	Description
list[str]	List of inactive user ID strings.

Source code in website/services/user.py

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

    Lightweight query that avoids loading full ORM objects.

    Returns:
        List of inactive user ID strings.
    """
    return self.repo.get_inactive_user_ids()

get_by_ids(ids)

Get 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/services/user.py

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

    Args:
        ids: List of user ID strings.

    Returns:
        List of User instances matching the given IDs.
    """
    return self.repo.get_by_ids(ids)

get_user_profile(user_id, force_refresh=False) staticmethod

Fetch a user's Discord profile.

Parameters:

Name	Type	Description	Default
user_id	str	Discord user ID.	required
force_refresh	bool	If True, bypass cache and fetch from Discord.	False

Returns:

Type	Description
dict	Dict with 'name', 'avatar', and optionally 'not_found' keys.

Source code in website/services/user.py

@staticmethod
def get_user_profile(user_id: str, force_refresh: bool = False) -> dict:
    """Fetch a user's Discord profile.

    Args:
        user_id: Discord user ID.
        force_refresh: If True, bypass cache and fetch from Discord.

    Returns:
        Dict with 'name', 'avatar', and optionally 'not_found' keys.
    """
    return _get_user_profile(user_id, force_refresh=force_refresh)

mark_inactive(user_id)

Mark a user as inactive (left the Discord server).

Parameters:

Name	Type	Description	Default
user_id	str	Discord user ID.	required

Returns:

Type	Description
User	Updated User instance.

Raises:

Type	Description
NotFoundError	If user does not exist.

Source code in website/services/user.py

def mark_inactive(self, user_id: str) -> User:
    """Mark a user as inactive (left the Discord server).

    Args:
        user_id: Discord user ID.

    Returns:
        Updated User instance.

    Raises:
        NotFoundError: If user does not exist.
    """
    user = self.get_by_id(user_id)
    user.not_player_as_of = datetime.now(timezone.utc)
    db.session.commit()
    return user

clear_inactive(user_id)

Clear the inactive flag for a user (they have rejoined).

Parameters:

Name	Type	Description	Default
user_id	str	Discord user ID.	required

Returns:

Type	Description
User	Updated User instance.

Raises:

Type	Description
NotFoundError	If user does not exist.

Source code in website/services/user.py

def clear_inactive(self, user_id: str) -> User:
    """Clear the inactive flag for a user (they have rejoined).

    Args:
        user_id: Discord user ID.

    Returns:
        Updated User instance.

    Raises:
        NotFoundError: If user does not exist.
    """
    user = self.get_by_id(user_id)
    user.not_player_as_of = None
    db.session.commit()
    return user

VttService

Service layer for Vtt (virtual tabletop) operations.

Handles CRUD operations with cache invalidation.

Source code in website/services/vtt.py

class VttService:
    """Service layer for Vtt (virtual tabletop) operations.

    Handles CRUD operations with cache invalidation.
    """

    def __init__(self, repository=None):
        self.repo = repository or VttRepository()

    @cache.memoize()
    def get_all(self) -> list[Vtt]:
        """Get all VTTs ordered by name.

        Returns:
            List of Vtt instances.
        """
        return self.repo.get_all_ordered()

    def get_by_id(self, id: int) -> Vtt:
        """Get VTT by ID.

        Args:
            id: VTT ID.

        Returns:
            Vtt instance.

        Raises:
            NotFoundError: If VTT does not exist.
        """
        vtt = self.repo.get_by_id(id)
        if not vtt:
            raise NotFoundError(
                f"Vtt with id {id} not found",
                resource_type="Vtt",
                resource_id=id,
            )
        return vtt

    def create(self, name: str, icon: str = None) -> Vtt:
        """Create a new VTT.

        Args:
            name: VTT name (must be unique).
            icon: Optional icon path.

        Returns:
            Created Vtt instance.

        Raises:
            ValidationError: If name already exists.
        """
        if self.repo.get_by_name(name):
            raise ValidationError("Vtt name already exists.", field="name")
        vtt = Vtt(name=name, icon=icon)
        self.repo.add(vtt)
        db.session.commit()
        cache.delete_memoized(self.get_all)
        return vtt

    def update(self, id: int, data: dict) -> Vtt:
        """Update an existing VTT.

        Args:
            id: VTT ID.
            data: Dictionary of fields to update.

        Returns:
            Updated Vtt instance.
        """
        vtt = self.repo.get_by_id_or_404(id)
        vtt.update_from_dict(data)
        db.session.commit()
        cache.delete_memoized(self.get_all)
        return vtt

    def delete(self, id: int) -> None:
        """Delete a VTT.

        Args:
            id: VTT ID.
        """
        vtt = self.repo.get_by_id_or_404(id)
        self.repo.delete(vtt)
        db.session.commit()
        cache.delete_memoized(self.get_all)

get_all()

Get all VTTs ordered by name.

Returns:

Type	Description
list[Vtt]	List of Vtt instances.

Source code in website/services/vtt.py

@cache.memoize()
def get_all(self) -> list[Vtt]:
    """Get all VTTs ordered by name.

    Returns:
        List of Vtt instances.
    """
    return self.repo.get_all_ordered()

get_by_id(id)

Get VTT by ID.

Parameters:

Name	Type	Description	Default
id	int	VTT ID.	required

Returns:

Type	Description
Vtt	Vtt instance.

Raises:

Type	Description
NotFoundError	If VTT does not exist.

Source code in website/services/vtt.py

def get_by_id(self, id: int) -> Vtt:
    """Get VTT by ID.

    Args:
        id: VTT ID.

    Returns:
        Vtt instance.

    Raises:
        NotFoundError: If VTT does not exist.
    """
    vtt = self.repo.get_by_id(id)
    if not vtt:
        raise NotFoundError(
            f"Vtt with id {id} not found",
            resource_type="Vtt",
            resource_id=id,
        )
    return vtt

create(name, icon=None)

Create a new VTT.

Parameters:

Name	Type	Description	Default
name	str	VTT name (must be unique).	required
icon	str	Optional icon path.	None

Returns:

Type	Description
Vtt	Created Vtt instance.

Raises:

Type	Description
ValidationError	If name already exists.

Source code in website/services/vtt.py

def create(self, name: str, icon: str = None) -> Vtt:
    """Create a new VTT.

    Args:
        name: VTT name (must be unique).
        icon: Optional icon path.

    Returns:
        Created Vtt instance.

    Raises:
        ValidationError: If name already exists.
    """
    if self.repo.get_by_name(name):
        raise ValidationError("Vtt name already exists.", field="name")
    vtt = Vtt(name=name, icon=icon)
    self.repo.add(vtt)
    db.session.commit()
    cache.delete_memoized(self.get_all)
    return vtt

update(id, data)

Update an existing VTT.

Parameters:

Name	Type	Description	Default
id	int	VTT ID.	required
data	dict	Dictionary of fields to update.	required

Returns:

Type	Description
Vtt	Updated Vtt instance.

Source code in website/services/vtt.py

def update(self, id: int, data: dict) -> Vtt:
    """Update an existing VTT.

    Args:
        id: VTT ID.
        data: Dictionary of fields to update.

    Returns:
        Updated Vtt instance.
    """
    vtt = self.repo.get_by_id_or_404(id)
    vtt.update_from_dict(data)
    db.session.commit()
    cache.delete_memoized(self.get_all)
    return vtt

delete(id)

Delete a VTT.

Parameters:

Name	Type	Description	Default
id	int	VTT ID.	required

Source code in website/services/vtt.py

def delete(self, id: int) -> None:
    """Delete a VTT.

    Args:
        id: VTT ID.
    """
    vtt = self.repo.get_by_id_or_404(id)
    self.repo.delete(vtt)
    db.session.commit()
    cache.delete_memoized(self.get_all)