feat(config): served from an in-process cache with a 5 s version check, writes bump the version atomically, mongo and postgres alike, redis keyed by version on top

copier.yml

@@ -71,7 +71,7 @@ taskiq_broker:
   choices:
     in-memory: memory
     Redis: redis
-    PostgreSQL (taskiq-pg):
+    PostgreSQL (taskiq-postgres):
       value: postgres
       validator: "{% if database != 'postgres' %}Requires the PostgreSQL database{% endif %}"
   default: memory

template/{{ part_dir }}/.env.example.jinja

@@ -1,7 +1,7 @@
 {%- set groups = [] -%}
 {%- if layout == 'root' %}{% set _ = groups.append(["COMPOSE_PROFILES=" ~ ("external," if database != 'none' or use_redis else "") ~ "services"]) %}{% endif -%}
 {%- set _ = groups.append(["RUN_ENVIRONMENT=prod", "TZ=UTC"]) -%}
-{%- set _ = groups.append(["LOG__LEVEL=INFO", "LOG__LEVEL_EXTERNAL=WARNING", "LOG__SHOW_TIME=false", "LOG__CONSOLE_WIDTH=150"]) -%}
+{%- set _ = groups.append(["LOG__LEVEL=INFO", "LOG__LEVEL_EXTERNAL=WARNING", "LOG__SHOW_TIME=false", "LOG__SHOW_LOCALS=false", "LOG__CONSOLE_WIDTH=150"]) -%}
 {%- if database == 'mongo' %}{% set _ = groups.append(["DB__HOST=mongodb", "DB__PORT=27017", "DB__USER=" ~ project_slug, "DB__PASSWORD=" ~ project_slug, "DB__DB_NAME=" ~ project_slug, "DB__SCRIPTS_CONNECTION_URL=mongodb://" ~ project_slug ~ ":" ~ project_slug ~ "@localhost:27017/" ~ project_slug ~ "?authSource=admin"]) %}{% endif -%}
 {%- if database == 'postgres' %}{% set _ = groups.append(["DB__HOST=postgres", "DB__PORT=5432", "DB__USER=" ~ project_slug, "DB__PASSWORD=" ~ project_slug, "DB__DB_NAME=" ~ project_slug, "DB__SCRIPTS_CONNECTION_URL=postgresql://" ~ project_slug ~ ":" ~ project_slug ~ "@localhost:5432/" ~ project_slug]) %}{% endif -%}
 {%- if use_redis %}{% set _ = groups.append(["REDIS__HOST=redis", "REDIS__PORT=6379", "REDIS__CACHE_TTL=300", "REDIS__SCRIPTS_CONNECTION_URL=redis://localhost:6379/0"]) %}{% endif -%}

template/{{ part_dir }}/pyproject.toml.jinja

@@ -4,7 +4,7 @@
 {%- if 'aiogram_bot' in backend_services %}{% set _ = deps.append("aiogram>=3.29.0") %}{% endif -%}
 {%- if include_payments %}{% set _ = deps.append("aiosend>=3.0.7") %}{% endif -%}
 {%- if 'kurigram_userbot' in backend_services %}{% set _ = deps.append("anyio>=4.0.0") %}{% set _ = deps.append("kurigram>=2.2.15") %}{% set _ = deps.append("tgcrypto>=1.2.5") %}{% set _ = deps.append("uvloop>=0.21.0") %}{% endif -%}
-{%- if 'taskiq_worker' in backend_services %}{% set _ = deps.append("taskiq>=0.12.4") %}{% if taskiq_broker == 'redis' %}{% set _ = deps.append("taskiq-redis>=1.2.3") %}{% endif %}{% if taskiq_broker == 'postgres' %}{% set _ = deps.append("taskiq-pg>=0.2.0") %}{% endif %}{% endif -%}
+{%- if 'taskiq_worker' in backend_services %}{% set _ = deps.append("taskiq>=0.13.0") %}{% if taskiq_broker == 'redis' %}{% set _ = deps.append("taskiq-redis>=1.2.3") %}{% endif %}{% if taskiq_broker == 'postgres' %}{% set _ = deps.append("taskiq-postgres[asyncpg]>=0.9.1") %}{% endif %}{% endif -%}
 {%- if 'scheduler' in backend_services %}{% set _ = deps.append("apscheduler>=3.11.0") %}{% endif -%}
 {%- if database == 'mongo' %}{% set _ = deps.append("beanie>=2.0.1") %}{% endif -%}
 {%- if database == 'postgres' %}{% set _ = deps.append("sqlmodel>=0.0.39") %}{% set _ = deps.append("asyncpg>=0.31.0") %}{% set _ = deps.append("alembic>=1.18.5") %}{% set _ = deps.append("greenlet>=3.5.3") %}{% endif -%}

template/{{ part_dir }}/scripts/{% if dynamic_config %}config{% endif %}/edit.py.jinja

@@ -261,10 +261,9 @@ async def read_config() -> DynamicConfigBase:
 
 async def write_config(config: DynamicConfigBase) -> DynamicConfigBase:
     await init_db()
-    doc = await DynamicConfig.get_or_create()
+    await DynamicConfig.update_config(
-    for name in DynamicConfigBase.model_fields:
+        **{name: getattr(config, name) for name in DynamicConfigBase.model_fields}
-        setattr(doc, name, getattr(config, name))
+    )
-    await doc.save()
     return config
 
 

template/{{ part_dir }}/src/utils/env.py.jinja

@@ -22,6 +22,7 @@ class LogSettings(Section):
     level: str = "INFO"
     level_external: str = "WARNING"
     show_time: bool = False
+    show_locals: bool = False
     console_width: int = 150
 {% if database == 'mongo' %}
 

template/{{ part_dir }}/src/utils/logging.py.jinja

@@ -28,7 +28,7 @@ def setup_logging() -> None:
                 markup=True,
                 rich_tracebacks=True,
                 enable_link_path=False,
-                tracebacks_show_locals=True,
+                tracebacks_show_locals=env.log.show_locals,
                 omit_repeated_times=False,
                 show_time=env.log.show_time,
                 {%- if 'aiogram_bot' in backend_services %}
@@ -37,7 +37,7 @@ def setup_logging() -> None:
             )
         ],
     )
-    install(console=console, show_locals=True)
+    install(console=console, show_locals=env.log.show_locals)
 
 
 logger = logging.getLogger("{{ project_slug }}")

template/{{ part_dir }}/src/utils/{% if database != 'none' %}db{% endif %}/models/{% if dynamic_config %}config.py{% endif %}.jinja

@@ -22,6 +22,7 @@ class DynamicConfig(SQLModel, table=True):
 
     id: int = SQLField(default=1, primary_key=True)
     data: dict[str, Any] = SQLField(sa_column=Column(postgresql.JSONB, nullable=False))
+    version: int = SQLField(default=0)
     updated_at: datetime | None = SQLField(
         default=None,
         sa_column=Column(
@@ -32,8 +33,19 @@ class DynamicConfig(SQLModel, table=True):
         ),
     )
 {%- else -%}
+import time
+from dataclasses import dataclass
+
 from beanie import Document
+from beanie.odm.operators.update import BaseUpdateOperator
+from beanie.operators import Inc, Set
 from pydantic import BaseModel, Field
+from pymongo.errors import PyMongoError
+
+from utils.logging import logger
+
+CONFIG_CHECK_SECONDS = 5.0
+CONFIG_TTL_SECONDS = 900.0
 
 
 class BotConfig(BaseModel):
@@ -44,15 +56,69 @@ class DynamicConfigBase(BaseModel):
     bot: BotConfig = Field(default_factory=BotConfig)
 
 
+@dataclass
+class _ConfigCache:
+    value: "DynamicConfig | None" = None
+    loaded_at: float = 0.0
+    checked_at: float = 0.0
+
+
+_cache = _ConfigCache()
+
+
 class DynamicConfig(DynamicConfigBase, Document):
+    version: int = 0
+
     class Settings:
         name = "config"
 
     @classmethod
     async def get_or_create(cls) -> "DynamicConfig":
-        config = await cls.find_one()
+        now = time.monotonic()
-        if config is None:
+        current = _cache.value
-            config = cls()
+        if current is None:
-            await config.save()
+            current = await cls._reload()
-        return config
+        elif now - _cache.loaded_at >= CONFIG_TTL_SECONDS:
+            current = await cls._reload_or_keep(current)
+        elif now - _cache.checked_at >= CONFIG_CHECK_SECONDS:
+            _cache.checked_at = now
+            current = await cls._refresh_if_changed(current)
+        return current.model_copy(deep=True)
+
+    @classmethod
+    async def update_config(
+        cls, *operations: BaseUpdateOperator, **sections: BaseModel
+    ) -> "DynamicConfig":
+        current = await cls.get_or_create()
+        changes = [*operations, Set(sections)] if sections else [*operations]
+        await cls.find_one(cls.id == current.id).update(*changes, Inc({cls.version: 1}))
+        return (await cls._reload()).model_copy(deep=True)
+
+    @classmethod
+    async def _reload(cls) -> "DynamicConfig":
+        loaded = await cls.find_one() or await cls().insert()
+        _cache.value = loaded
+        _cache.loaded_at = _cache.checked_at = time.monotonic()
+        return loaded
+
+    @classmethod
+    async def _reload_or_keep(cls, current: "DynamicConfig") -> "DynamicConfig":
+        try:
+            return await cls._reload()
+        except PyMongoError as e:
+            logger.warning(f"Config reload failed, serving the cached copy: {e}")
+            return current
+
+    @classmethod
+    async def _refresh_if_changed(cls, current: "DynamicConfig") -> "DynamicConfig":
+        try:
+            stored = await cls.get_pymongo_collection().find_one(
+                {"_id": current.id}, {"version": 1}
+            )
+        except PyMongoError as e:
+            logger.warning(f"Config version check failed, serving the cached copy: {e}")
+            return current
+        if stored and stored.get("version", 0) == current.version:
+            return current
+        return await cls._reload_or_keep(current)
 {%- endif %}

template/{{ part_dir }}/src/utils/{% if database != 'none' %}db{% endif %}/{% if database == 'postgres' %}repositories{% endif %}/{% if dynamic_config %}config.py{% endif %}.jinja

@@ -1,88 +1,139 @@
-{% if use_redis -%}
+import time
+{%- if use_redis %}
+from contextlib import suppress
+{%- endif %}
+from typing import Any
+
+from pydantic import BaseModel
+{%- if use_redis %}
+from redis.exceptions import RedisError
+{%- endif %}
+from sqlalchemy import literal, update
+from sqlalchemy.dialects.postgresql import JSONB
+from sqlalchemy.exc import SQLAlchemyError
 from sqlalchemy.ext.asyncio import async_sessionmaker
+from sqlmodel import col, select
 from sqlmodel.ext.asyncio.session import AsyncSession
 
 from utils.db.models.config import DynamicConfig, DynamicConfigBase
+from utils.logging import logger
+{%- if use_redis %}
 
 from .cache import RedisCache
+{%- endif %}
 
-CONFIG_CACHE_KEY = "config:dynamic"
 CONFIG_ROW_ID = 1
+CONFIG_CHECK_SECONDS = 5.0
+CONFIG_TTL_SECONDS = 900.0
+{%- if use_redis %}
+CONFIG_CACHE_KEY = "config:dynamic"
+{%- endif %}
 
 
 class ConfigRepository:
     def __init__(
-        self, sessionmaker: async_sessionmaker[AsyncSession], cache: RedisCache
+        self, sessionmaker: async_sessionmaker[AsyncSession]{% if use_redis %}, cache: RedisCache{% endif %}
     ) -> None:
         self._sessionmaker = sessionmaker
+{%- if use_redis %}
         self._cache = cache
-
-    async def get(self) -> DynamicConfigBase:
-        return await self._cache.get_or_set_model(
-            CONFIG_CACHE_KEY, DynamicConfigBase, self._load
-        )
-
-    async def _load(self) -> DynamicConfigBase:
-        async with self._sessionmaker() as session:
-            row = await session.get(DynamicConfig, CONFIG_ROW_ID)
-            if row is None:
-                config = DynamicConfigBase()
-                session.add(DynamicConfig(id=CONFIG_ROW_ID, data=config.model_dump()))
-                await session.commit()
-                return config
-            return DynamicConfigBase.model_validate(row.data)
-
-    async def save(self, config: DynamicConfigBase) -> DynamicConfigBase:
-        async with self._sessionmaker() as session:
-            row = await session.get(DynamicConfig, CONFIG_ROW_ID)
-            if row is None:
-                session.add(DynamicConfig(id=CONFIG_ROW_ID, data=config.model_dump()))
-            else:
-                row.data = config.model_dump()
-                session.add(row)
-            await session.commit()
-        await self._cache.set_model(CONFIG_CACHE_KEY, config)
-        return config
-
-    async def reset(self) -> DynamicConfigBase:
-        return await self.save(DynamicConfigBase())
-
-    async def invalidate(self) -> None:
-        await self._cache.invalidate(CONFIG_CACHE_KEY)
-{%- else -%}
-from sqlalchemy.ext.asyncio import async_sessionmaker
-from sqlmodel.ext.asyncio.session import AsyncSession
-
-from utils.db.models.config import DynamicConfig, DynamicConfigBase
-
-CONFIG_ROW_ID = 1
-
-
-class ConfigRepository:
-    def __init__(self, sessionmaker: async_sessionmaker[AsyncSession]) -> None:
-        self._sessionmaker = sessionmaker
-
-    async def get(self) -> DynamicConfigBase:
-        async with self._sessionmaker() as session:
-            row = await session.get(DynamicConfig, CONFIG_ROW_ID)
-            if row is None:
-                config = DynamicConfigBase()
-                session.add(DynamicConfig(id=CONFIG_ROW_ID, data=config.model_dump()))
-                await session.commit()
-                return config
-            return DynamicConfigBase.model_validate(row.data)
-
-    async def save(self, config: DynamicConfigBase) -> DynamicConfigBase:
-        async with self._sessionmaker() as session:
-            row = await session.get(DynamicConfig, CONFIG_ROW_ID)
-            if row is None:
-                session.add(DynamicConfig(id=CONFIG_ROW_ID, data=config.model_dump()))
-            else:
-                row.data = config.model_dump()
-                session.add(row)
-            await session.commit()
-        return config
-
-    async def reset(self) -> DynamicConfigBase:
-        return await self.save(DynamicConfigBase())
 {%- endif %}
+        self._value: DynamicConfigBase | None = None
+        self._version = -1
+        self._loaded_at = 0.0
+        self._checked_at = 0.0
+
+    async def get(self) -> DynamicConfigBase:
+        now = time.monotonic()
+        current = self._value
+        if current is None:
+            current = await self._reload()
+        elif now - self._loaded_at >= CONFIG_TTL_SECONDS:
+            current = await self._reload_or_keep(current)
+        elif now - self._checked_at >= CONFIG_CHECK_SECONDS:
+            self._checked_at = now
+            current = await self._refresh_if_changed(current)
+        return current.model_copy(deep=True)
+
+    async def save(self, config: DynamicConfigBase) -> DynamicConfigBase:
+        return await self._write(config.model_dump(mode="json"), merge=False)
+
+    async def update(self, **sections: BaseModel) -> DynamicConfigBase:
+        patch = {name: value.model_dump(mode="json") for name, value in sections.items()}
+        return await self._write(patch, merge=True)
+
+    async def reset(self) -> DynamicConfigBase:
+        return await self.save(DynamicConfigBase())
+
+    async def _write(self, data: dict[str, Any], *, merge: bool) -> DynamicConfigBase:
+        await self.get()
+        payload = literal(data, JSONB)
+        async with self._sessionmaker() as session:
+            await session.exec(
+                update(DynamicConfig)
+                .where(col(DynamicConfig.id) == CONFIG_ROW_ID)
+                .values(
+                    data=col(DynamicConfig.data).op("||")(payload) if merge else payload,
+                    version=col(DynamicConfig.version) + 1,
+                )
+            )
+            await session.commit()
+        return (await self._reload()).model_copy(deep=True)
+
+    async def _stored_version(self) -> int | None:
+        async with self._sessionmaker() as session:
+            query = select(DynamicConfig.version).where(
+                col(DynamicConfig.id) == CONFIG_ROW_ID
+            )
+            return (await session.exec(query)).first()
+
+    async def _reload(self) -> DynamicConfigBase:
+{%- if use_redis %}
+        version = await self._stored_version()
+        if version is not None:
+            with suppress(RedisError):
+                key = f"{CONFIG_CACHE_KEY}:{version}"
+                cached = await self._cache.get_model(key, DynamicConfigBase)
+                if cached is not None:
+                    return self._remember(cached, version)
+{%- endif %}
+        async with self._sessionmaker() as session:
+            row = await session.get(DynamicConfig, CONFIG_ROW_ID)
+            if row is None:
+                row = DynamicConfig(
+                    id=CONFIG_ROW_ID, data=DynamicConfigBase().model_dump(mode="json")
+                )
+                session.add(row)
+                await session.commit()
+                await session.refresh(row)
+            value = DynamicConfigBase.model_validate(row.data)
+            version = row.version
+{%- if use_redis %}
+        with suppress(RedisError):
+            await self._cache.set_model(f"{CONFIG_CACHE_KEY}:{version}", value)
+{%- endif %}
+        return self._remember(value, version)
+
+    def _remember(self, value: DynamicConfigBase, version: int) -> DynamicConfigBase:
+        self._value, self._version = value, version
+        self._loaded_at = self._checked_at = time.monotonic()
+        return value
+
+    async def _reload_or_keep(self, current: DynamicConfigBase) -> DynamicConfigBase:
+        try:
+            return await self._reload()
+        except SQLAlchemyError as e:
+            logger.warning(f"Config reload failed, serving the cached copy: {e}")
+            return current
+
+    async def _refresh_if_changed(
+        self, current: DynamicConfigBase
+    ) -> DynamicConfigBase:
+        try:
+            version = await self._stored_version()
+        except SQLAlchemyError as e:
+            logger.warning(f"Config version check failed, serving the cached copy: {e}")
+            return current
+        if version == self._version:
+            return current
+        return await self._reload_or_keep(current)

template/{{ part_dir }}/src/{% if 'taskiq_worker' in backend_services %}worker{% endif %}/broker.py.jinja

@@ -43,7 +43,7 @@ setup_dishka(container, broker)
 {% if use_dishka %}from dishka.integrations.taskiq import setup_dishka
 {% endif %}{% if 'scheduler' in backend_services %}from taskiq import TaskiqScheduler
 from taskiq.schedule_sources import LabelScheduleSource
-{% endif %}from taskiq_pg import AsyncpgBroker, AsyncpgResultBackend
+{% endif %}from taskiq_pg.asyncpg import AsyncpgBroker, AsyncpgResultBackend
 
 {% if use_dishka %}from dependencies.container import container
 {% endif %}from utils.env import env

template/{{ part_dir }}/{% if database == 'postgres' %}migrations{% endif %}/env.py

@@ -16,10 +16,15 @@ if config.config_file_name is not None:
 target_metadata = SQLModel.metadata
 
 
+def include_name(name: str | None, type_: str, _parents: object) -> bool:
+    return not (type_ == "table" and name is not None and name.startswith("taskiq_"))
+
+
 def run_migrations_offline() -> None:
     context.configure(
         url=env.db.connection_url,
         target_metadata=target_metadata,
+        include_name=include_name,
         literal_binds=True,
         dialect_opts={"paramstyle": "named"},
     )
@@ -28,7 +33,11 @@ def run_migrations_offline() -> None:
 
 
 def do_run_migrations(connection: Connection) -> None:
-    context.configure(connection=connection, target_metadata=target_metadata)
+    context.configure(
+        connection=connection,
+        target_metadata=target_metadata,
+        include_name=include_name,
+    )
     with context.begin_transaction():
         context.run_migrations()
 

template/{{ part_dir }}/{% if layout == 'part' %}CLAUDE.md{% endif %}

@@ -11,3 +11,6 @@ Every service imports only what it needs to start. `make check` runs `scripts/ch
 - Package `__init__` files don't re-export clients or SDKs.
 - An SDK used by one function is imported inside it. On an async path load it first with `await preload("<module>")` from `utils.imports`, so the event loop never waits on the import.
 - Types used in Dishka provider signatures and `FromDishka[...]` stay real top-level imports.
+
+## Dynamic config
+`config()` returns a copy of an in-process snapshot: no database round-trip per call, changes from any process show up within 5 s through a version number on the config record. Never `save()` a config you read; write through `DynamicConfig.update_config(section=...)` (Mongo) or `ConfigRepository.update(section=...)`/`save(...)` (Postgres), which bump the version atomically. Counters and other hot data don't belong in the config, give them their own collection or table with atomic increments.