Skip to content

refactor: fix Sonar type hints and nested contexts #625

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
ReenigneArcher merged 1 commit into from
Oct 4, 2026
Merged

refactor: fix Sonar type hints and nested contexts #625

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 1 addition & 2 deletions src/common/database.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,6 @@
import shelve
import shutil
import threading
from typing import Union

# lib imports
import git
Expand All @@ -25,7 +24,7 @@


class Database:
def __init__(self, db_name: str, db_dir: Union[str, Path] = data_dir, use_git: bool = True):
def __init__(self, db_name: str, db_dir: str | Path = data_dir, use_git: bool = True):
self.db_name = db_name
self.db_dir = db_dir
self.repo = None
Expand Down
60 changes: 30 additions & 30 deletions src/common/rank.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,7 @@
import random
import threading
import time
from typing import Dict, List, Optional, Tuple, Union
from typing import Dict, List, Optional, Tuple

# lib imports
import aiohttp
Expand Down Expand Up @@ -60,20 +60,20 @@ def __init__(
self.last_activity = {} # Tracks last activity time for cooldowns

@staticmethod
def get_community_id(platform: str, user: Union[DiscordUser, RedditUser]) -> Optional[Union[int, str]]:
def get_community_id(platform: str, user: DiscordUser | RedditUser) -> int | str | None:
"""
Get the community ID for a user based on platform.

Parameters
----------
platform : str
Platform identifier ('discord' or 'reddit')
user : Union[DiscordUser, RedditUser]
user : DiscordUser | RedditUser
Discord or Reddit user object

Returns
-------
Union[int, str]
int | str
Community ID (guild_id for Discord, subreddit_id for Reddit)
"""
if platform == 'discord':
Expand Down Expand Up @@ -138,7 +138,7 @@ def calculate_xp_for_level(level: int) -> int:
def get_rank_data(
self,
platform: str,
user: Union[DiscordUser, RedditUser],
user: DiscordUser | RedditUser,
create_if_not_exists: bool = False,
) -> dict:
"""
Expand All @@ -148,7 +148,7 @@ def get_rank_data(
----------
platform : str
Platform identifier (e.g., 'discord', 'reddit')
user : Union[DiscordUser, RedditUser]
user : DiscordUser | RedditUser
Discord or Reddit user object
create_if_not_exists : bool
If True, create a new user entry if it doesn't exist
Expand Down Expand Up @@ -189,15 +189,15 @@ def get_rank_data(

return user_data

def update_rank_data(self, platform: str, user: Union[DiscordUser, RedditUser], data: dict) -> dict:
def update_rank_data(self, platform: str, user: DiscordUser | RedditUser, data: dict) -> dict:
"""
Update rank data for a user.

Parameters
----------
platform : str
Platform identifier (e.g., 'discord', 'reddit')
user : Union[DiscordUser, RedditUser]
user : DiscordUser | RedditUser
Discord or Reddit user object
data : dict
New data to update
Expand All @@ -219,15 +219,15 @@ def update_rank_data(self, platform: str, user: Union[DiscordUser, RedditUser],
data=data,
)

def award_xp(self, platform: str, user: Union[DiscordUser, RedditUser]) -> Optional[dict]:
def award_xp(self, platform: str, user: DiscordUser | RedditUser) -> Optional[dict]:
"""
Award XP to a user with cooldown enforcement.

Parameters
----------
platform : str
Platform identifier (e.g., 'discord', 'reddit')
user : Union[DiscordUser, RedditUser]
user : DiscordUser | RedditUser
Discord or Reddit user object

Returns
Expand Down Expand Up @@ -287,8 +287,8 @@ def award_xp(self, platform: str, user: Union[DiscordUser, RedditUser]) -> Optio
def get_leaderboard(
self,
platform: str,
user: Optional[Union[DiscordUser, RedditUser]] = None,
community_id: Optional[Union[int, str]] = None,
user: DiscordUser | RedditUser | None = None,
community_id: int | str | None = None,
limit: int = 100,
offset: int = 0,
) -> List[dict]:
Expand All @@ -299,9 +299,9 @@ def get_leaderboard(
----------
platform : str
Platform identifier (e.g., 'discord', 'reddit')
user : Optional[Union[DiscordUser, RedditUser]]
user : DiscordUser | RedditUser | None
User object to determine community_id if not explicitly provided
community_id : Optional[Union[int, str]]
community_id : int | str | None
Community identifier (guild_id for Discord, subreddit_id for Reddit)
limit : int
Maximum number of entries to return
Expand Down Expand Up @@ -338,15 +338,15 @@ def get_leaderboard(

return leaderboard

def get_user_rank_position(self, platform: str, user: Union[DiscordUser, RedditUser]) -> Optional[int]:
def get_user_rank_position(self, platform: str, user: DiscordUser | RedditUser) -> Optional[int]:
"""
Get the exact rank position of a user on the leaderboard without any limits.

Parameters
----------
platform : str
Platform identifier (e.g., 'discord', 'reddit')
user : Union[DiscordUser, RedditUser]
user : DiscordUser | RedditUser
Discord or Reddit user object

Returns
Expand Down Expand Up @@ -377,8 +377,8 @@ def get_user_rank_position(self, platform: str, user: Union[DiscordUser, RedditU
def get_migration_status(
self,
platform: str,
community_id: Union[int, str],
source_id: Union[int, str],
community_id: int | str,
source_id: int | str,
) -> Optional[dict]:
"""
Check if migration has already been performed for a specific source.
Expand All @@ -387,9 +387,9 @@ def get_migration_status(
----------
platform : str
Platform identifier (e.g., 'discord', 'reddit')
community_id : Union[int, str]
community_id : int | str
Community identifier (guild_id or subreddit_id)
source_id : Union[int, str]
source_id : int | str
Source identifier (e.g., guild_id for Mee6)

Returns
Expand All @@ -406,8 +406,8 @@ def get_migration_status(
def set_migration_completed(
self,
platform: str,
community_id: Union[int, str],
source_id: Union[int, str],
community_id: int | str,
source_id: int | str,
stats: dict,
) -> dict:
"""
Expand All @@ -417,9 +417,9 @@ def set_migration_completed(
----------
platform : str
Platform identifier (e.g., 'discord', 'reddit')
community_id : Union[int, str]
community_id : int | str
Community identifier (guild_id or subreddit_id)
source_id : Union[int, str]
source_id : int | str
Source identifier (e.g., guild_id for Mee6)
stats : dict
Migration statistics
Expand All @@ -441,7 +441,7 @@ def migrate_from_reddit_database(
reddit_bot,
reddit_db,
community_id: str,
) -> Dict[str, Union[int, str]]:
) -> Dict[str, int | str]:
"""
Migrate user data from Reddit database.

Expand All @@ -458,7 +458,7 @@ def migrate_from_reddit_database(

Returns
-------
Dict[str, Union[int, str]]
Dict[str, int | str]
Migration statistics
"""
# Acquire migration lock to prevent concurrent migrations
Expand Down Expand Up @@ -600,7 +600,7 @@ def _do_reddit_migration(
reddit_bot,
reddit_db,
community_id: str,
) -> Dict[str, Union[int, str]]:
) -> Dict[str, int | str]:
"""Internal method that performs the actual Reddit migration."""
total_submissions = 0
total_comments = 0
Expand Down Expand Up @@ -681,7 +681,7 @@ def _do_reddit_migration(
logger.info(f"Reddit migration completed with stats: {stats}")
return stats

async def migrate_from_mee6(self, guild_id: int) -> Dict[str, Union[int, str]]:
async def migrate_from_mee6(self, guild_id: int) -> Dict[str, int | str]:
"""
Migrate user data from Mee6 API.

Expand All @@ -692,7 +692,7 @@ async def migrate_from_mee6(self, guild_id: int) -> Dict[str, Union[int, str]]:

Returns
-------
Dict[str, Union[int, str]]
Dict[str, int | str]
Migration statistics
"""
import asyncio
Expand Down Expand Up @@ -818,7 +818,7 @@ def _process_mee6_batch(

return new_users, updated_users

async def _do_mee6_migration(self, guild_id: int) -> Dict[str, Union[int, str]]:
async def _do_mee6_migration(self, guild_id: int) -> Dict[str, int | str]:
"""Internal method that performs the actual Mee6 migration."""
page = 0
total_users = 0
Expand Down
42 changes: 21 additions & 21 deletions src/common/rank_database.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
# standard imports
from typing import List, Optional, Union
from typing import List, Optional

# local imports
from src.common.database import Database
Expand Down Expand Up @@ -39,7 +39,7 @@ def _ensure_tables(self):
def get_community_users(
self,
platform: str,
community_id: Union[int, str],
community_id: int | str,
search: Optional[str] = None,
) -> List[dict]:
"""
Expand All @@ -49,7 +49,7 @@ def get_community_users(
----------
platform : str
Platform identifier ('discord' or 'reddit')
community_id : Union[int, str]
community_id : int | str
Community identifier (guild_id for Discord, subreddit_id for Reddit)
search : Optional[str]
Optional search string to filter users by username (case-insensitive)
Expand Down Expand Up @@ -84,8 +84,8 @@ def get_community_users(
def get_user_data(
self,
platform: str,
community_id: Union[int, str],
user_id: Union[int, str],
community_id: int | str,
user_id: int | str,
create_if_not_exists: bool = False,
) -> dict:
"""
Expand All @@ -95,9 +95,9 @@ def get_user_data(
----------
platform : str
Platform identifier ('discord' or 'reddit')
community_id : Union[int, str]
community_id : int | str
Community identifier (guild_id for Discord, subreddit_id for Reddit)
user_id : Union[int, str]
user_id : int | str
User identifier
create_if_not_exists : bool
Whether to create a new user entry if it doesn't exist
Expand Down Expand Up @@ -140,8 +140,8 @@ def get_user_data(
def update_user_data(
self,
platform: str,
community_id: Union[int, str],
user_id: Union[int, str],
community_id: int | str,
user_id: int | str,
data: dict,
) -> dict:
"""
Expand All @@ -151,9 +151,9 @@ def update_user_data(
----------
platform : str
Platform identifier ('discord' or 'reddit')
community_id : Union[int, str]
community_id : int | str
Community identifier (guild_id or subreddit_id)
user_id : Union[int, str]
user_id : int | str
User identifier
data : dict
New user data
Expand Down Expand Up @@ -196,7 +196,7 @@ def update_user_data(
def get_leaderboard(
self,
platform: str,
community_id: Union[int, str],
community_id: int | str,
limit: int = 100,
offset: int = 0,
) -> List[dict]:
Expand All @@ -207,7 +207,7 @@ def get_leaderboard(
----------
platform : str
Platform identifier ('discord' or 'reddit')
community_id : Union[int, str]
community_id : int | str
Community identifier (guild_id or subreddit_id)
limit : int
Maximum number of entries to return
Expand Down Expand Up @@ -251,8 +251,8 @@ def get_leaderboard(
def get_migration_status(
self,
platform: str,
community_id: Union[int, str],
source_id: Union[int, str],
community_id: int | str,
source_id: int | str,
) -> Optional[dict]:
"""
Check if migration has already been performed for a specific source.
Expand All @@ -261,9 +261,9 @@ def get_migration_status(
----------
platform : str
Platform identifier (e.g., 'discord', 'reddit')
community_id : Union[int, str]
community_id : int | str
Community identifier (guild_id or subreddit_id)
source_id : Union[int, str]
source_id : int | str
Source identifier (e.g., guild_id for Mee6)

Returns
Expand All @@ -285,8 +285,8 @@ def get_migration_status(

def set_migration_completed(
self, platform: str,
community_id: Union[int, str],
source_id: Union[int, str],
community_id: int | str,
source_id: int | str,
stats: dict,
) -> dict:
"""
Expand All @@ -296,9 +296,9 @@ def set_migration_completed(
----------
platform : str
Platform identifier ('discord' or 'reddit')
community_id : Union[int, str]
community_id : int | str
Community identifier (guild_id or subreddit_id)
source_id : Union[int, str]
source_id : int | str
Source identifier (e.g., guild_id for Mee6)
stats : dict
Migration statistics
Expand Down
Loading
Loading