Metadata-Version: 2.4
Name: totvs-dta-utils
Version: 1.5.1
Summary: Lib for integration with DTA services
Author: TOTVS - IDEIA
Author-email: info@totvs.ai
Requires-Python: >=3.10,<4.0
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Provides-Extra: apikeys
Provides-Extra: audit
Provides-Extra: auth
Provides-Extra: cloud-tasks
Provides-Extra: error-handling
Provides-Extra: healthcheck
Provides-Extra: pubsub
Provides-Extra: secrets
Requires-Dist: PyJWT (>=2.0.0) ; extra == "auth"
Requires-Dist: cryptography (>=3.4.0) ; extra == "auth"
Requires-Dist: dta-observability (>=0.0.19) ; extra == "audit"
Requires-Dist: fastapi (>=0.136.1,<1.0) ; extra == "error-handling" or extra == "healthcheck"
Requires-Dist: google-cloud-pubsub (>=2.0.0,<3.0.0) ; extra == "pubsub"
Requires-Dist: google-cloud-tasks (>=2.0.0) ; extra == "cloud-tasks"
Requires-Dist: httpx (>=0.23.0) ; extra == "healthcheck"
Requires-Dist: pydantic (>=2.0.0)
Requires-Dist: pydantic-settings (>=2.0.0)
Requires-Dist: python-dotenv (>=1.0.0) ; extra == "secrets" or extra == "apikeys"
Requires-Dist: redis (>=4.0.0) ; extra == "auth"
Requires-Dist: requests (>=2.0.0)
Description-Content-Type: text/markdown

# Dta Utils 🧰 🛠️
**Agilize a integração entre serviços DTA**


### O que são Serviços DTA?

Uma coleção de serviços para facilitar e acelerar o desenvolvimento e monitoramento de Aplicações, com foco em aplicativos de IA generativa.


## Introdução

Esse pacote possui módulos extras que auxiliam o desenvolvimento de integrações com os serviços do DTA.

## Extra "Cloud Tasks"

Abstração para criação de HTTP tasks no Google Cloud Tasks, centralizando a comunicação e evitando acoplamento direto à lib `google-cloud-tasks` nas aplicações consumidoras.

### Instalação

```shell
pip install "totvs-dta-utils[cloud_tasks]"
```

Ou utilizando `poetry`:
```shell
poetry add "totvs-dta-utils[cloud_tasks]"
```

### Utilização genérica

```python
from dta_utils_python.cloud_tasks import CloudTasksClient, CloudTasksConfig, HttpTaskRequest

client = CloudTasksClient(
    CloudTasksConfig(
        project_id="my-project",
        location_id="us-central1",
        queue_id="my-queue",
    )
)

task_name = client.create_http_task(
    HttpTaskRequest(
        url="https://my-service/internal/some-job",
        payload={"key": "value"},
        task_name="my-deterministic-task",  # opcional — garante idempotência
    )
)
```

### Retenção por tenant

```python
from dta_utils_python.cloud_tasks import CloudTasksClient, CloudTasksConfig
from dta_utils_python.cloud_tasks.retention import create_retention_task_for_tenant

client = CloudTasksClient(
    CloudTasksConfig(
        project_id="my-project",
        location_id="us-central1",
        queue_id="retention-queue",
    )
)

# Idempotente: chamar duas vezes no mesmo dia/tenant não gera erro
task_name = create_retention_task_for_tenant(
    client=client,
    worker_url="https://my-service/internal/retention/process-tenant",
    tenant_schema="tenant-a",
    retention_days=180,
    batch_size=500,
)
```

O nome da task gerado segue o padrão `retention-{YYYYMMDD}-{tenant_schema}`, com caracteres especiais sanitizados.

### Autenticação OIDC

Para endpoints protegidos, o Cloud Tasks pode gerar e anexar um OIDC token automaticamente na entrega da task. Basta informar o service account e, opcionalmente, o audience (padrão: a própria URL da task):

```python
from dta_utils_python.cloud_tasks import CloudTasksClient, CloudTasksConfig, HttpTaskRequest, OidcConfig

client = CloudTasksClient(
    CloudTasksConfig(
        project_id="my-project",
        location_id="us-central1",
        queue_id="my-queue",
    )
)

task_name = client.create_http_task(
    HttpTaskRequest(
        url="https://my-service/internal/some-job",
        payload={"key": "value"},
        oidc_token=OidcConfig(
            service_account_email="my-sa@my-project.iam.gserviceaccount.com",
            # audience="https://my-service/internal/some-job",  # opcional
        ),
    )
)
```

### Emulator local

Configure via variável de ambiente:

```env
CLOUD_TASKS_EMULATOR_HOST=localhost:8123
```

Ou via configuração:

```python
CloudTasksConfig(
    project_id="local-project",
    location_id="us-central1",
    queue_id="my-queue",
    emulator_host="localhost:8123",
)
```

---

## Extra "Secrets"

### Instalação

Instale o módulo `secrets` com:
```shell
pip install "totvs-dta-utils[secrets]"
```

Ou utilizando `poetry`:
```shell
poetry add "totvs-dta-utils[secrets]"
```

### Configuração inicial:

Adicione as seguintes variaveis no `.env` do seu projeto:
```env
DTA_ENVIRONMENT="development"
DTA_INTEGRATION_URL="{DTA_INTEGRATION_URL}"
```
> NOTE: Para ambiente em cloud, onde terá acesso irrestrito aos secrets, o valor do `DTA_ENVIRONMENT`deve ser `production`.

### Utilização

```python
from dta_utils_python import DtaSecrets

auth = DTA_JWT  # CLIENT AUTHORIZATION

secrets = DtaSecrets(authorization=auth,
                     project="dta-empodera")

all_secrets = secrets.all()  # Get the latest version of all secrets
my_secret = secrets.get("MY_SECRET")  # Get the latest version of a secret
my_secret_v2 = secrets.get("MY_SECRET", version=2)  # Get a specific version of a secret
```
> Observação: Para ambiente em nuvem na rede DTA, nenhuma autenticação é necessária.

> Observação 2: Ainda em ambientes de nuvem, usando Cloud Run, lembrar de habilitar TODAS as chamadas de saída do serviço DEVEM passar pela VPC. Selecione `Route all traffic to the VPC` na configuração de Rede do serviço Cloud Run

### Demais configurações:
```python
DtaSecrets(
    authorization=auth,
    project="dta-empodera",
    raise_exception: bool = True,  # Default "False" - Levanta exceção em caso de erro ao obter a secret
    autoload: bool = False,  # Default "True" - Pré-carrega todas as secrets do projeto na inicialização da classe e as mantém em cache de memória
)
```

### Tipos de retorno:
- `.get("SECRET_2")`:
Retorna o valor da secret ou `None` caso a secret não exista.
```python
any: "321654"
```

- `.all()`:
Retorna um dicionário (hashmap) contendo a última versão de todas as secrets
```python
dict: {
    "SECRET_1": "123456",
    "SECRET_2": "321654",
    "SECRET_3": "My secret",
}
```

## Extra "Auth" (Identity/JWT)

### Desenvolvimento local (Identity)

Para viabilizar autorização via **Identity** em ambiente local (sem Redis/RAC), configure:

```env
DTA_ENVIRONMENT="development"  # ou "local"
DTA_TENANT_NAME="dta-alisson"
DTA_TOTVS_TENANT_ID="1a2b3c-4d5e6f"
DTA_ROLE_USER="TENANT_ADMIN"
# Opcional: sobrescreve o endpoint JWKS usado para validar o JWT Identity
# DTA_IDENTITY_JWKS_BASE_URL="https://api-fluig.totvs.app/accounts/api/v1/jwks"
```

Com isso, `get_identity_user_from_cache(...)` passa a:
- Validar o JWT Identity via JWKS (RS256)
- Ignorar Redis (cache) para o token
- Resolver `dta_roles` a partir de `DTA_*`

## Extra "Healthcheck"

Componente reutilizável de health check para aplicações FastAPI. Expõe endpoints padronizados de liveness e readiness.

### Instalação

```shell
pip install "totvs-dta-utils[healthcheck]"
```

### Endpoints

| Endpoint | Método | Descrição |
|---|---|---|
| `/health/live` | GET | Liveness — processo está rodando |
| `/health/ready` | GET | Readiness — dependências externas OK |

### Utilização

```python
from fastapi import FastAPI
from dta_utils_python.healthcheck import create_health_router, DtaHealthCheck

app = FastAPI()

async def check_database() -> bool:
    # await db.execute("SELECT 1")
    return True

def check_redis() -> bool:
    # redis_client.ping()
    return True

app.include_router(
    create_health_router(
        readiness_checks=[
            DtaHealthCheck(name="database", check=check_database),
            DtaHealthCheck(name="redis", check=check_redis),
        ]
    )
)
```

### Respostas

**`GET /health/live`** — sempre 200:
```json
{"status": "ok"}
```

**`GET /health/ready`** — 200 se todos os checks passarem, 503 caso contrário:
```json
{
  "status": "ok",
  "checks": {
    "database": "ok",
    "redis": "ok"
  }
}
```

### Configurações opcionais

```python
create_health_router(
    readiness_checks=[...],
    prefix="/api/health",   # default: "/health"
    live_path="/ping",      # default: "/live"
    ready_path="/status",   # default: "/ready"
)
```

```python
DtaHealthCheck(
    name="database",
    check=check_database,
    timeout=5.0,  # segundos; default: 2.0
)
```

> Checks sync e async são suportados. Exceções e timeouts são tratados como falha sem vazar detalhes na resposta.

