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
This commit is contained in:
hh committed
1 parent
4cb3b85321
commit
eed732fbcf
4 files changed
+200
-81
No files matched your search
+71
-5
@@ -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()
|
||||
if config is None:
|
||||
config = cls()
|
||||
await config.save()
|
||||
return config
|
||||
now = time.monotonic()
|
||||
current = _cache.value
|
||||
if current is None:
|
||||
current = await cls._reload()
|
||||
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 %}
|
||||
+123
-72
@@ -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)
|
||||
Reference in new issue
Block a user