🪛Use toml config file instead

.gitignore

@@ -11,7 +11,7 @@ venv/
 
 *.ini
 
-config.py
+config.toml
 
 .temp/
 

.vscode/settings.json

@@ -23,6 +23,7 @@
         "ibot",
         "irebot",
         "ireloop",
+        "IRENESTEST",
         "kmmrbot",
         "lolrankbot",
         "loremipsum",

__main__.py

@@ -2,13 +2,15 @@
 
 import asyncio
 import logging
+import platform
 import sys
 
 import aiohttp
+import asyncpg
 import click
 
 from bot import IreBot, setup_logging
-from utils.database import create_pool
+from config import config
 
 try:
     import uvloop  # pyright: ignore[reportMissingImports]
@@ -20,6 +22,18 @@
     asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
 
 
+async def create_pool() -> asyncpg.Pool[asyncpg.Record]:
+    """Create AsyncPG Pool."""
+    postgres_url = config["POSTGRES"]["VPS"] if platform.system() == "Linux" else config["POSTGRES"]["HOME"]
+    return await asyncpg.create_pool(
+        postgres_url,
+        command_timeout=60,
+        min_size=10,
+        max_size=10,
+        statement_cache_size=0,
+    )  # type: ignore[reportReturnType]
+
+
 async def start_the_bot() -> None:
     """Start the bot."""
     log = logging.getLogger()

bot/bot.py

@@ -2,6 +2,7 @@
 
 import asyncio
 import logging
+import platform
 from typing import TYPE_CHECKING, TypedDict, override
 
 import discord
@@ -10,7 +11,7 @@
 from twitchio.ext import commands
 
 # from twitchio.web import StarletteAdapter
-import config
+from config import config, replace_secrets
 from ext import EXTENSIONS
 from ext.dota.api import Dota2Client
 from utils import const, errors
@@ -22,7 +23,7 @@
     import asyncpg
     from aiohttp import ClientSession
 
-    from utils.database import PoolTypedWithAny
+    from types_.database import PoolTypedWithAny
 
     class LoadTokensQueryRow(TypedDict):
         user_id: str
@@ -78,8 +79,8 @@ def __init__(
         #     eventsub_secret=config.EVENTSUB_SECRET,
         # )
         super().__init__(
-            client_id=config.TTV_DEV_CLIENT_ID,
-            client_secret=config.TTV_DEV_CLIENT_SECRET,
+            client_id=config["TWITCH"]["CLIENT_ID"],
+            client_secret=config["TWITCH"]["CLIENT_SECRET"],
             bot_id=const.UserID.Bot,
             owner_id=const.UserID.Irene,
             prefix=self.prefixes,
@@ -332,7 +333,7 @@ async def event_command_error(self, payload: commands.CommandErrorPayload) -> No
             #     await ctx.send(str(error))
 
             case _:
-                await ctx.send(f"{error.__class__.__name__}: {config.replace_secrets(str(error))}")
+                await ctx.send(f"{error.__class__.__name__}: {replace_secrets(str(error))}")
 
                 command_name = getattr(ctx.command, "name", "unknown")
 
@@ -377,27 +378,27 @@ def webhook_from_url(self, url: str) -> discord.Webhook:
     @discord.utils.cached_property
     def logger_webhook(self) -> discord.Webhook:
         """A webhook in hideout's #logger channel."""
-        return self.webhook_from_url(config.LOGGER_WEBHOOK)
+        return self.webhook_from_url(config["WEBHOOKS"]["LOGGER"])
 
     @discord.utils.cached_property
     def error_webhook(self) -> discord.Webhook:
         """A webhook in hideout server to send errors/notifications to the developer(-s)."""
-        return self.webhook_from_url(config.ERROR_WEBHOOK)
+        return self.webhook_from_url(config["WEBHOOKS"]["ERROR"])
 
-    @property
+    @discord.utils.cached_property
     def error_ping(self) -> str:
         """Error Role ping used to notify the developer(-s) about some errors."""
-        return config.ERROR_PING
+        return "<@&1306631362870120600>" if platform.system() == "Windows" else "<@&1116171071528374394>"
 
-    async def aluerie_stream(self) -> twitchio.Stream | None:
-        """Shortcut to get @Aluerie's stream."""
+    async def irene_stream(self) -> twitchio.Stream | None:
+        """Shortcut to get @Irene's stream."""
         return next(iter(await self.fetch_streams(user_ids=[const.UserID.Irene])), None)
 
     @ireloop(count=1)
     async def check_if_online(self) -> None:
         """Check if aluerie is online - used to make my own (proper) online event instead of twitchio's."""
         await asyncio.sleep(1.0)  # just in case;
-        if await self.aluerie_stream():
+        if await self.irene_stream():
             self.irene_online = True
             self.dispatch("aluerie_online")
 

bot/logs.py

@@ -128,7 +128,6 @@ def get_log_fmt(handler: logging.Handler) -> logging.Formatter:
         formatter = logging.Formatter(
             "%(asctime)s %(levelname)-4.4s %(name)-30s %(lineno)-4d %(funcName)-35s %(message)s",
             "%H:%M:%S %d/%m",
-            style="{",
         )
 
     return formatter

config.py

@@ -0,0 +1,33 @@
+from __future__ import annotations
+
+import re
+import tomllib
+from pathlib import Path
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from types_.config import Config
+
+__all__ = (
+    "config",
+    "replace_secrets",
+)
+
+
+with Path("config.toml").open("rb") as fp:
+    config: Config = tomllib.load(fp)  # pyright: ignore[reportAssignmentType]
+
+
+DO_NOT_SPOIL_LIST: list[str] = [
+    config["TWITCH"]["CLIENT_ID"],
+    config["TWITCH"]["CLIENT_SECRET"],
+    config["TOKENS"]["STEAM"],
+]
+
+
+SPOIL_PATTERN = re.compile("|".join(map(re.escape, DO_NOT_SPOIL_LIST)))
+
+
+def replace_secrets(text: str) -> str:
+    """Hide Secrets from my config files from a string."""
+    return SPOIL_PATTERN.sub("SECRET", text)

examples/beta/base.py

@@ -9,8 +9,8 @@
 import twitchio
 from twitchio.ext import commands
 
-import config
 from bot import IreComponent, ireloop
+from config import config
 from utils import const
 
 if TYPE_CHECKING:

ext/dota/api/dota2client.py

@@ -1,12 +1,13 @@
 from __future__ import annotations
 
 import logging
+import platform
 from typing import TYPE_CHECKING, override
 
 from steam import PersonaState
 from steam.ext.dota2 import Client
 
-import config
+from config import config
 
 from .pulsefire_clients import SteamWebAPIClient, StratzClient
 from .storage import Items
@@ -37,7 +38,7 @@ def __init__(self, twitch_bot: IreBot) -> None:
         self.items = Items(twitch_bot)
 
     def irene(self) -> PartialUser:
-        return self.instantiate_partial_user(config.IRENE_STEAM_ID64)
+        return self.instantiate_partial_user(config["STEAM"]["IRENE_ID64"])
 
     async def start_helpers(self) -> None:
         if not self.started:
@@ -47,7 +48,11 @@ async def start_helpers(self) -> None:
 
     @override
     async def login(self) -> None:
-        await super().login(config.STEAM_USERNAME, config.STEAM_PASSWORD)
+        account_credentials = (
+            config["STEAM"]["IRENESTEST"] if platform.system() == "Windows" else config["STEAM"]["IRENESBOT"]
+        )
+        username, password = account_credentials["USERNAME"], account_credentials["PASSWORD"]
+        await super().login(username, password)
 
     @override
     async def close(self) -> None:

ext/dota/api/pulsefire_clients.py

@@ -11,7 +11,7 @@
 from pulsefire.middlewares import http_error_middleware, json_response_middleware
 from pulsefire.ratelimiters import BaseRateLimiter
 
-import config
+from config import config
 
 if TYPE_CHECKING:
     from collections.abc import Mapping, Sequence
@@ -40,7 +40,7 @@ def __init__(self) -> None:
             base_url="https://api.steampowered.com/",
             default_params={},
             default_headers={},
-            default_queries={"key": config.STEAM_WEB_API_KEY},
+            default_queries={"key": config["TOKENS"]["STEAM"]},
             middlewares=[
                 json_response_middleware(orjson.loads),
                 http_error_middleware(),
@@ -173,7 +173,7 @@ def __init__(self) -> None:
             default_params={},
             default_headers={
                 "User-Agent": "STRATZ_API",
-                "Authorization": f"Bearer {config.STRATZ_BEARER_TOKEN}",
+                "Authorization": f"Bearer {config['TOKENS']['STRATZ_BEARER']}",
                 "Content-Type": "application/json",
             },
             default_queries={},

ext/dota/commands.py

@@ -7,8 +7,8 @@
 
 from twitchio.ext import commands
 
-import config
 from bot import IreComponent, ireloop
+from config import config
 from utils import const, errors, helpers
 
 from .models import Streamer
@@ -39,7 +39,7 @@ class DotaCommands(IreComponent):
 
     def __init__(self, bot: IreBot) -> None:
         super().__init__(bot)
-        self.streamer: Streamer = Streamer(self.bot, config.IRENE_STEAM_ID64)
+        self.streamer: Streamer = Streamer(self.bot, config["STEAM"]["IRENE_ID64"])
         self.debug_mode: bool = True
 
     async def debug_send(self, message: str) -> None:

ext/simple_commands.py

@@ -9,8 +9,8 @@
 import twitchio  # noqa: TC002
 from twitchio.ext import commands
 
-import config
 from bot import IreComponent
+from config import config
 from utils import const, formats, guards
 
 if TYPE_CHECKING:
@@ -239,7 +239,7 @@ async def shoutout(self, ctx: commands.Context, user: twitchio.User) -> None:
     @commands.command()
     async def song(self, ctx: commands.Context) -> None:
         """Get currently played song on Spotify."""
-        url = f"https://spotify.aidenwallis.co.uk/u/{config.SPOTIFY_AIDENWALLIS_CODE}"
+        url = f"https://spotify.aidenwallis.co.uk/u/{config['TOKENS']['SPOTIFY_AIDENWALLIS']}"
         async with self.bot.session.get(url) as resp:
             msg = await resp.text()
 
@@ -264,7 +264,7 @@ async def source(self, ctx: commands.Context) -> None:
     @commands.command()
     async def uptime(self, ctx: commands.Context) -> None:
         """Get stream uptime."""
-        stream = await self.bot.aluerie_stream()
+        stream = await self.bot.irene_stream()
         if stream is None:
             await ctx.send(f"Stream is offline {const.BTTV.Offline}")
         else:

types_/config.py

@@ -0,0 +1,46 @@
+from typing import TypedDict
+
+__all__ = ("Config",)
+
+
+class Twitch(TypedDict):
+    CLIENT_ID: str
+    CLIENT_SECRET: str
+
+
+class Postgres(TypedDict):
+    VPS: str
+    HOME: str
+
+
+class SteamAccount(TypedDict):
+    USERNAME: str
+    PASSWORD: str
+
+
+class Steam(TypedDict):
+    IRENE_ID64: int
+    IRENESTEST: SteamAccount
+    IRENESBOT: SteamAccount
+
+
+class Tokens(TypedDict):
+    STRATZ_BEARER: str
+    STEAM: str
+    SPOTIFY_AIDENWALLIS: str
+    EVENTSUB: str
+
+
+class Webhooks(TypedDict):
+    ERROR: str
+    LOGGER: str
+
+
+class Config(TypedDict):
+    """Type-hints for dictionary created from loading `config.toml` file."""
+
+    POSTGRES: Postgres
+    STEAM: Steam
+    TWITCH: Twitch
+    WEBHOOKS: Webhooks
+    TOKENS: Tokens

types_/database.py

@@ -0,0 +1,26 @@
+from typing import Any, override
+
+import asyncpg
+
+
+class PoolTypedWithAny(asyncpg.Pool[asyncpg.Record]):
+    """Fake Type Class.
+
+    For typing purposes, our `bot.pool` will be "type-ignore"'d-as `PoolTypedWithAny`
+    that allows us to properly type the return values via narrowing like mentioned in instructions above
+    without hundreds of "type: ignore" notices for each TypedDict.
+
+    I could use Protocol to type it all, but `async-stubs` provide a good job in typing most of the stuff
+    and we also don't lose doc-string this way.
+
+    * Right now, asyncpg is untyped so this is better than the current status quo
+    * If we ever need the regular Pool type we have `bot.database` without any shenanigans.
+    """
+
+    # all methods below were changed from "asyncpg.Record" to "Any"
+
+    @override
+    async def fetch(self, query: str, *args: Any, timeout: float | None = None) -> list[Any]: ...
+
+    @override
+    async def fetchrow(self, query: str, *args: Any, timeout: float | None = None) -> Any: ...