---

Instalação pública

O WatchGuardian está disponível publicamente no PyPI.

pip install watchguardian

Para atualizar para a versão mais recente:

pip install --upgrade watchguardian

Depois da instalação, os imports públicos são:

from WatchGuardian import Settings, WatchLogs, WatchObs, WatchAuth
from WatchGuardian.database import WatchDb, CacheManage
from WatchGuardian.server import ClientHttp, WatchServer

E a CLI fica disponível diretamente no terminal:

watchguardian --help

---

O que é o WatchGuardian?

O WatchGuardian é um framework, não apenas uma coleção de helpers.

Ele fornece uma camada integrada para aplicações Python que precisam de observabilidade, persistência, autenticação, cache e comunicação entre serviços sem reconstruir a mesma infraestrutura em cada projeto.

O framework pode ser usado de duas formas:

1. Embutido na própria aplicação, usando módulos como WatchLogs, WatchObs, WatchDb, CacheManage e WatchAuth.
2. Como servidor central de observabilidade, usando WatchServer para subir uma API própria e ClientHttp para outras aplicações se conectarem a ela.

Principais recursos

• logging com DEBUG, INFO, SUCCESS, WARNING, ERROR e CRITICAL;
• persistência opcional de logs em background;
• observabilidade assíncrona de tarefas, latência, status e erros;
• servidor próprio de observabilidade baseado em FastAPI;
• PostgreSQL assíncrono e SQLite como fallback;
• Redis para cache, TTL, contadores, hashes e Pub/Sub;
• JWT com user_token e refresh_token;
• cliente HTTP para comunicação com instâncias WatchGuardian;
• CLI própria com watchguardian;
• testes unitários, de integração e funcionais;
• CI automatizado com GitHub Actions;
• distribuição pública via PyPI;
• projeto open source sob licença MIT, aberto a sugestões e contribuições via Pull Request.

---

Servidor próprio de observabilidade

Um dos pontos centrais do framework é a possibilidade de montar uma instância WatchGuardian dedicada à observabilidade.

Aplicação A ─┐
Aplicação B ─┼──> WatchGuardian Server ───> Logs / Observabilidade / Banco / Redis
Aplicação C ─┘

Essa instância centraliza:

• logs de múltiplos serviços;
• tarefas e status de execução;
• latência;
• falhas e erros;
• autenticação das requisições;
• persistência de dados;
• consultas remotas via ClientHttp;
• rate limit e cache com Redis.

Assim, o WatchGuardian pode funcionar tanto como framework dentro de um projeto quanto como um serviço separado de observabilidade para vários sistemas.

---

Arquitetura

flowchart TD
    A[Aplicação] --> B[WatchGuardian API pública]

    B --> C[Core]
    B --> D[Database]
    B --> E[Server]

    C --> F[Settings]
    C --> G[WatchLogs]
    C --> H[WatchObs]
    C --> I[WatchAuth]

    D --> J[WatchDb]
    D --> K[CacheManage]

    E --> L[ClientHttp]
    E --> M[WatchServer]

    M --> N[Servidor de observabilidade]
    N --> O[Logs]
    N --> P[Observabilidade]
    N --> Q[Autenticação]
    N --> R[Redis]
    N --> S[Database]

A camada WatchGuardian/ é a API pública. A implementação interna permanece em src/, permitindo evoluir a arquitetura sem obrigar quem usa o framework a depender da estrutura interna.

---

Uso dos módulos

Settings

Centraliza variáveis de ambiente e configurações.

from WatchGuardian import Settings

settings = Settings(env_file=".env")

settings.add_secret([
    "secret",
    "host",
    "port",
])

secret = settings.required_secrets("secret")
host = settings.get_secret("host")

Configurações em memória também podem ser registradas:

settings.add_config({
    "name": "environment",
    "value": "development",
})

mode = settings.required_config("environment")

---

WatchLogs

Gerencia logs estruturados.

from WatchGuardian import WatchLogs

logs = WatchLogs(loglevel="DEBUG")

logs.debug("Entrando no fluxo")
logs.info("Processando dados")
logs.success("Operação concluída")
logs.warning("Tempo de resposta alto")
logs.error("Falha na operação")
logs.critical("Falha crítica")

Para persistir os logs:

from WatchGuardian import WatchLogs
from WatchGuardian.database import WatchDb

watch_db = await WatchDb(settings=settings)
logs = WatchLogs(loglevel="INFO", watch_db=watch_db)

logs.info("Esse log também será persistido")

Quando watch_db é informado, a persistência é executada em background para não bloquear o fluxo principal.

---

WatchObs

Registra execução, status, latência e falhas de tarefas.

from WatchGuardian import WatchLogs, WatchObs
from WatchGuardian.database import WatchDb

watch_db = await WatchDb(settings=settings)
logs = WatchLogs(loglevel="INFO", watch_db=watch_db)
obs = WatchObs(logs=logs, watch_db=watch_db)

async with obs.begin("process-data"):
    await process_data()

Ao entrar no contexto a tarefa é registrada. Ao sair, o framework finaliza automaticamente a observabilidade, incluindo erros quando ocorrerem.

---

WatchAuth

Cria e valida JWTs.

from WatchGuardian import Settings, WatchAuth

settings = Settings()
settings.add_secret("secret")

auth = WatchAuth(settings=settings)

user_token = await auth.user_token({"user_id": "123"})
refresh_token = await auth.refresh_token({"user_id": "123"})

payload = await auth.decode(user_token)

Gerar uma secret:

from WatchGuardian import WatchAuth

secret = WatchAuth().secret()
print(secret)

---

WatchDb

Fornece acesso ao banco usado pelo framework e aos dados de logs e observabilidade.

from WatchGuardian import Settings
from WatchGuardian.database import WatchDb

settings = Settings()
settings.add_secret("url_db")

watch_db = await WatchDb(settings=settings)

Sem url_db, o framework pode utilizar SQLite como fallback.

Tabelas

await watch_db.create_tables()
await watch_db.reset_tables()
await watch_db.drop_tables()

Logs

created = await watch_db.logs.create(
    log="Aplicação iniciada",
    status="INFO",
)

logs = await watch_db.logs.select()

Observabilidade

await watch_db.observability.create(
    task="etl",
    status="pending",
    content="starting",
    latency=0,
)

---

CacheManage

Interface Redis do framework.

from WatchGuardian import Settings
from WatchGuardian.database import CacheManage

settings = Settings()
settings.add_secret(["host", "port", "password"])

cache = CacheManage(settings=settings)

Contador com TTL:

value = await cache.incr("rate:user:123", time=60)

Hash:

await cache.hash(
    key="user:123",
    data={"name": "Brayan", "status": "active"},
    time=3600,
)

user = await cache.read("user:123", hash=True)

Pub/Sub:

await cache.publish("watchguardian:events", "task_finished")

async for event in cache.listen("watchguardian:events"):
    print(event)

---

WatchServer

Monta o servidor próprio de observabilidade do framework.

from WatchGuardian import Settings
from WatchGuardian.server import WatchServer

settings = Settings()
settings.add_secret([
    "secret",
    "origin",
    "host",
    "port",
    "password",
    "rate_limit",
])

server = WatchServer(settings=settings)
app = server.run()

Depois execute com Uvicorn:

uvicorn main:app --reload

O servidor inclui rotas internas de logs, observabilidade e autenticação, além de CORS, rate limit e integração com Redis.

---

ClientHttp

Conecta aplicações ao servidor WatchGuardian.

from WatchGuardian import Settings
from WatchGuardian.server import ClientHttp

settings = Settings()
client = ClientHttp(settings=settings)

Por padrão usa http://localhost:8000. Também pode usar a variável url.

logs = await client.logs(token=user_token).get_log()
items = await client.obs(token=user_token).get_obs()

result = await client.auth(
    user_token=user_token,
    refresh_token=refresh_token,
).refresh()

---

CLI

Após instalar pelo PyPI:

pip install watchguardian

A CLI fica disponível como:

watchguardian [comando]

Ajuda

watchguardian --help

Banco

watchguardian migrate make_tables
watchguardian migrate reset
watchguardian migrate drop

Com arquivo .env específico:

watchguardian migrate --env-file .env.production make_tables

Autenticação

watchguardian auth secret
watchguardian auth token --type user_token --payload '{"user_id":"123"}'
watchguardian auth token --type refresh_token --payload '{"user_id":"123"}'

Servidor

watchguardian server run main:app

watchguardian server run main:app --host 0.0.0.0 --port 8080

watchguardian server run main:app --detached

Consultar dados do servidor

watchguardian server logs --token SEU_TOKEN
watchguardian server obs --token SEU_TOKEN

---

Exemplo rápido da versão pública

Depois de instalar:

pip install watchguardian

Crie um arquivo main.py:

import asyncio
import os

from WatchGuardian import Settings, WatchLogs, WatchObs, WatchAuth
from WatchGuardian.database import WatchDb


async def main() -> None:
    os.environ["secret"] = WatchAuth().secret()

    settings = Settings()
    settings.add_secret("secret")

    db = await WatchDb(settings=settings)
    await db.create_tables()

    logs = WatchLogs(loglevel="INFO", watch_db=db)
    obs = WatchObs(logs=logs, watch_db=db)
    auth = WatchAuth(settings=settings)

    token = await auth.user_token({"user_id": "123"})

    logs.success("WatchGuardian iniciado")

    async with obs.begin("example-task"):
        await asyncio.sleep(0.2)

    print(token)


asyncio.run(main())

Execute:

python main.py

---

Variáveis de ambiente

# JWT
secret=your-secret

# Redis
host=localhost
port=6379
password=

# Database
url_db=postgresql+asyncpg://user:password@localhost:5432/watchguardian

# Server
origin=http://localhost:3000
rate_limit=100

# Client HTTP
url=http://localhost:8000

Cada módulo utiliza apenas as configurações necessárias para sua função.

---

Testes e CI/CD

O projeto possui três níveis de testes:

Unitários

Validam componentes isolados como banco, logs, cache, token e comandos internos.

Integração

Validam a comunicação entre módulos, incluindo integração entre banco, logging e observabilidade.

Funcionais

Sobem o servidor WatchGuardian e exercitam o fluxo completo via cliente HTTP e CLI.

Os testes são executados no GitHub Actions. O workflow build.yaml executa testes unitários e de integração com Redis como serviço, enquanto server.yaml executa os testes funcionais do servidor.

O projeto também está preparado para fluxo de CI/CD com GitHub Actions e publicação de releases no PyPI. Atualmente a validação contínua é automatizada e a distribuição pública é feita pelo pacote watchguardian.

---

Open source e contribuições

O WatchGuardian é open source e o desenvolvimento é aberto à comunidade.

Sugestões, melhorias, correções de bugs, documentação e novos recursos podem ser enviados por Pull Request no GitHub. Antes de abrir um PR, prefira manter a API pública compatível, adicionar ou atualizar testes quando necessário e garantir que os workflows de CI continuem passando.

Fluxo recomendado para contribuir:

1. faça um fork do repositório;
2. crie uma branch para a alteração;
3. implemente a mudança e execute os testes;
4. faça push da branch;
5. abra um Pull Request descrevendo o problema, a solução e o impacto da mudança.

Issues também podem ser usadas para propor recursos, relatar bugs ou discutir mudanças antes da implementação.

---

Licença

O WatchGuardian é distribuído sob a MIT License. Isso permite usar, copiar, modificar, distribuir e incorporar o framework em projetos pessoais ou comerciais, respeitando os termos presentes no arquivo LICENSE.

---

Estrutura do projeto

WatchGuardian/
├── WatchGuardian/          # API pública do framework
│   ├── __init__.py
│   ├── database.py
│   └── server.py
│
├── src/                    # implementação interna
│   ├── auth/
│   ├── cache/
│   ├── cli/
│   ├── client/
│   ├── core/
│   ├── database/
│   ├── logs/
│   ├── server/
│   ├── service/
│   └── util/
│
├── tests/
│   ├── unit/
│   ├── integration/
│   └── functional/
│
├── .github/workflows/
├── assets/
├── LICENSE
├── pyproject.toml
└── README.md

---

Desenvolvimento local

git clone https://github.com/BrayanDevZN/WatchGuardian.git
cd WatchGuardian
python -m venv .venv
source .venv/bin/activate
pip install -e .

Gerar o pacote:

python -m pip install build
python -m build

---

Status

Versão pública atual: 0.1.0.

• Framework instalável via PyPI
• API pública em WatchGuardian
• Settings
• Logging
• Observabilidade
• JWT
• Banco assíncrono
• Redis
• HTTP Client
• Servidor próprio de observabilidade
• CLI watchguardian
• Testes unitários
• Testes de integração
• Testes funcionais
• CI com GitHub Actions
• Open source sob licença MIT

---