metricallize 0.1.0

TinyDB-based document storage for simple telemetry/logging with sync, cached and serialized async modes.

metricallize

Uma forma simples de armazenar telemetria/logs em “banco de documentos” usando TinyDB, com 3 modos de persistência:

• readable_sync: JSON legível (escreve no disco de forma direta)
• in_memory_async: JSON legível com cache em memória (escreve em batch, no flush/close)
• serialized_async: arquivo binário/comprimido (mais rápido/menor, escreve em background; exige flush/close)

Instalação

pip install metricallize

Para usar o modo serialized_async (serialização/compressão):

pip install "metricallize[serialized_async]"

Conceitos

• Você cria um DocumentStorage
• Você acessa qualquer table com storage.db.table("nome")
• O decorator storage.capture(table="...") registra uma “execução” (status/duração/erro) nessa table

Campos automáticos gravados pelo capture:

• ts: YYYY-MM-DD HH:MM:SS.xxx (UTC)
• name: module.qualname da função
• status: ok ou error
• duration: HH:MM:SS
• Em caso de erro: exc_type, exc_msg

Path e versionamento (por mês)

Se você passar path como diretório (sem extensão), o arquivo é criado automaticamente em:

<path>/storage/YYYY_MM/day_DD.json (modos legíveis)
<path>/storage/YYYY_MM/day_DD.db (modo serialized)

Você pode controlar o nome base com name="...".

Uso: readable_sync (JSON legível)

from pathlib import Path
from metricallize import DocumentStorage

storage_dir = Path.cwd() / "data"
storage = DocumentStorage(path=storage_dir, mode="readable_sync", name="logger")

logger = storage.db.table("loggers")
logger.insert({"event": "startup"})

Uso: in_memory_async (cache em memória, JSON legível)

from pathlib import Path
from metricallize import DocumentStorage

storage_dir = Path.cwd() / "data"
storage = DocumentStorage(
    path=storage_dir,
    mode="in_memory_async",
    name="trace",
    write_cache_size=100_000,
)

trace = storage.db.table("trace")
trace.insert({"event": "step"})

storage.flush()
storage.close()

Uso: serialized_async (binário, assíncrono)

from pathlib import Path
from metricallize import DocumentStorage

storage_dir = Path.cwd() / "data"
storage = DocumentStorage(path=storage_dir, mode="serialized_async", name="trace")

trace = storage.db.table("trace")
trace.insert({"event": "step"})

storage.flush()
storage.close()

Decorator: capture

from pathlib import Path
from metricallize import DocumentStorage

storage = DocumentStorage(path=Path.cwd() / "data", mode="readable_sync")
table_name = "trace"

@storage.capture(table=table_name)
def soma(a: int, b: int) -> int:
    return a + b

@storage.capture(table=table_name)
def falha() -> None:
    raise RuntimeError("boom")

soma(1, 2)
try:
    falha()
except RuntimeError:
    pass

Publicação (TestPyPI e PyPI)

Build:

python -m pip install --upgrade build twine
python -m build

TestPyPI:

python -m twine upload --repository testpypi dist/*

PyPI:

python -m twine upload dist/*