Compare commits

..

No commits in common. "main" and "QA-01" have entirely different histories.
main ... QA-01

970 changed files with 8368 additions and 26641 deletions

View file

@ -1,9 +1,13 @@
# Usa a imagem oficial do Python
FROM python:3.12-slim FROM python:3.12-slim
# Define diretório de trabalho no container
WORKDIR /app WORKDIR /app
# Copia o arquivo de dependências
COPY requirements.txt . COPY requirements.txt .
# Instala dependências no sistema e no Python
RUN apt-get update && apt-get install -y \ RUN apt-get update && apt-get install -y \
gcc libffi-dev libssl-dev python3-dev firebird-dev \ gcc libffi-dev libssl-dev python3-dev firebird-dev \
&& pip install --upgrade pip \ && pip install --upgrade pip \
@ -12,13 +16,11 @@ RUN apt-get update && apt-get install -y \
&& apt-get autoremove -y \ && apt-get autoremove -y \
&& rm -rf /var/lib/apt/lists/* && rm -rf /var/lib/apt/lists/*
# =============================== # Copia o restante do projeto para o container
# CRIA DIRETÓRIOS NECESSÁRIOS
# ===============================
RUN mkdir -p storage/temp
COPY . . COPY . .
# Expõe a porta padrão do Uvicorn/FastAPI
EXPOSE 8000 EXPOSE 8000
CMD ["gunicorn", "main:app", "-w", "4", "-k", "uvicorn.workers.UvicornWorker", "--bind", "0.0.0.0:8000"] # Comando para iniciar o servidor
CMD ["sh", "-c", "uvicorn main:app --host 0.0.0.0 --port 8000"]

View file

@ -70,7 +70,7 @@
"header": [], "header": [],
"body": { "body": {
"mode": "raw", "mode": "raw",
"raw": "{\r\n \"login\" : \"Kenio\",\r\n \"senha_api\": \"123123\"\r\n}", "raw": "{\r\n \"login\" : \"Kenio0\",\r\n \"senha_api\": \"123123\"\r\n}",
"options": { "options": {
"raw": { "raw": {
"language": "json" "language": "json"
@ -8501,81 +8501,6 @@
}, },
"response": [] "response": []
}, },
{
"name": "GET GRUPO",
"event": [
{
"listen": "test",
"script": {
"exec": [
"pm.test(\"Status code é 201 ou 200\", function () {\r",
" pm.expect(pm.response.code).to.be.oneOf([200, 201]);\r",
"});\r",
"\r",
"pm.test(\"Resposta contém ID ou confirmação\", function () {\r",
" const jsonData = pm.response.json();\r",
" pm.expect(jsonData).to.have.any.keys(\"id\", \"message\", \"status\");\r",
"});\r",
"\r",
"pm.test(\"Mensagem indica sucesso\", function () {\r",
" const jsonData = pm.response.json();\r",
" pm.expect(jsonData.message || \"\").to.match(/sucesso|criado|registrado/i);\r",
"});\r",
""
],
"type": "text/javascript",
"packages": {},
"requests": {}
}
}
],
"protocolProfileBehavior": {
"disableBodyPruning": true
},
"request": {
"auth": {
"type": "bearer",
"bearer": [
{
"key": "token",
"value": "{{BearerToken}}",
"type": "string"
}
]
},
"method": "GET",
"header": [],
"body": {
"mode": "raw",
"raw": ""
},
"url": {
"raw": "{{BaseUrlV1}}administrativo/g_marcacao_tipo/grupo?grupo=MODELO_ETIQUETA&sistema_id=2&situacao=A",
"host": [
"{{BaseUrlV1}}administrativo"
],
"path": [
"g_marcacao_tipo",
"grupo"
],
"query": [
{
"key": "grupo",
"value": "MODELO_ETIQUETA"
},
{
"key": "sistema_id",
"value": "2"
},
{
"key": "situacao",
"value": "A"
}
]
}
},
"response": []
},
{ {
"name": "GET ID", "name": "GET ID",
"event": [ "event": [
@ -8816,65 +8741,6 @@
} }
] ]
}, },
{
"name": "Emolumentos",
"item": [
{
"name": "GET",
"event": [
{
"listen": "test",
"script": {
"exec": [
"pm.test(\"Status code é 201 ou 200\", function () {\r",
" pm.expect(pm.response.code).to.be.oneOf([200, 201]);\r",
"});\r",
"\r",
"pm.test(\"Resposta contém ID ou confirmação\", function () {\r",
" const jsonData = pm.response.json();\r",
" pm.expect(jsonData).to.have.any.keys(\"id\", \"message\", \"status\");\r",
"});\r",
"\r",
"pm.test(\"Mensagem indica sucesso\", function () {\r",
" const jsonData = pm.response.json();\r",
" pm.expect(jsonData.message || \"\").to.match(/sucesso|criado|registrado/i);\r",
"});\r",
""
],
"type": "text/javascript",
"packages": {},
"requests": {}
}
}
],
"request": {
"auth": {
"type": "bearer",
"bearer": [
{
"key": "token",
"value": "{{BearerToken}}",
"type": "string"
}
]
},
"method": "GET",
"header": [],
"url": {
"raw": "{{BaseUrlV1}}administrativo/g_emolumento/1",
"host": [
"{{BaseUrlV1}}administrativo"
],
"path": [
"g_emolumento",
"1"
]
}
},
"response": []
}
]
},
{ {
"name": "Andamento Serviço", "name": "Andamento Serviço",
"item": [ "item": [

240
README.md
View file

@ -1,12 +1,12 @@
# Configuração do Projeto Python # Configuração do Projeto Python
Este guia descreve o passo a passo para configurar o ambiente de desenvolvimento e produção de um projeto Python, incluindo ambiente virtual, dependências, banco de dados, e ajuste de desempenho com múltiplos núcleos. Este guia descreve o passo a passo para configurar o ambiente de desenvolvimento de um projeto Python, incluindo a preparação do ambiente virtual, instalação de dependências e configuração do banco de dados.
--- ---
## 1. Clonar o Projeto ## 1. Clonar o Projeto
Clone o repositório do projeto a partir do Git: Primeiro, clone o repositório do projeto a partir do Git:
```bash ```bash
git clone https://git.oriustecnologia.com/OriusTecnologia/saas_api.git git clone https://git.oriustecnologia.com/OriusTecnologia/saas_api.git
@ -16,7 +16,7 @@ git clone https://git.oriustecnologia.com/OriusTecnologia/saas_api.git
## 2. Criar o Ambiente Virtual ## 2. Criar o Ambiente Virtual
Crie um **ambiente virtual** isolado para o projeto: O uso de um **ambiente virtual** garante que as bibliotecas instaladas para este projeto não afetem o Python global da sua máquina.
```bash ```bash
python -m venv venv python -m venv venv
@ -26,13 +26,14 @@ python -m venv venv
## 3. Ativar o Ambiente Virtual ## 3. Ativar o Ambiente Virtual
Ative o ambiente virtual antes de instalar as dependências: Ative o ambiente virtual antes de instalar as dependências ou executar a aplicação.
```bash ```bash
venv\Scripts\activate venv\Scripts\activate
``` ```
> **Em sistemas Linux/Mac:** > **Observação:**
> Em sistemas Unix (Linux/Mac), o comando pode ser:
> >
> ```bash > ```bash
> source venv/bin/activate > source venv/bin/activate
@ -42,11 +43,9 @@ venv\Scripts\activate
## 4. Instalar Dependências do Sistema ## 4. Instalar Dependências do Sistema
O projeto depende de compiladores nativos para algumas bibliotecas Python. A biblioteca de criptografia utilizada no projeto requer uma extensão da Microsoft para ser instalada.
Baixe e instale o **Microsoft C++ Build Tools** através do link abaixo:
### Windows
Baixe e instale o **Microsoft C++ Build Tools**:
[https://visualstudio.microsoft.com/pt-br/visual-cpp-build-tools/](https://visualstudio.microsoft.com/pt-br/visual-cpp-build-tools/) [https://visualstudio.microsoft.com/pt-br/visual-cpp-build-tools/](https://visualstudio.microsoft.com/pt-br/visual-cpp-build-tools/)
Durante a instalação, selecione o pacote: Durante a instalação, selecione o pacote:
@ -55,15 +54,6 @@ Durante a instalação, selecione o pacote:
Desktop Development With C++ Desktop Development With C++
``` ```
### Linux
Execute no terminal:
```bash
sudo apt update
sudo apt install -y build-essential libpq-dev
```
--- ---
## 5. Instalar as Bibliotecas do Projeto ## 5. Instalar as Bibliotecas do Projeto
@ -76,189 +66,77 @@ pip install -r requirements.txt
--- ---
## 6. Configuração do Banco de Dados (Firebird) ## 6. Configurar o Banco de Dados
Crie o arquivo **`.env`** na raiz do projeto e configure as variáveis de ambiente necessárias para a conexão com o banco de dados Firebird: O projeto utiliza um banco **Firebird**.
Edite o arquivo de configuração localizado em:
```env ```
ORIUS_API_FDB_HOST=localhost api/config/database/firebird.json
ORIUS_API_FDB_NAME=S:/Bases/SANTARITA.FDB
ORIUS_API_FDB_PORT=3050
ORIUS_API_FDB_USER=SYSDBA
ORIUS_API_FDB_PASSWORD=302b3c
ORIUS_API_FDB_CHARSET=UTF8
ORIUS_API_FDB_POOL_PRE_PING=true
ORIUS_API_FDB_POOL_SIZE=5
ORIUS_API_FDB_POOL_MAX_OVERFLOW=10
ORIUS_CLIENT_STATE=go
``` ```
Essas configurações definem o acesso ao banco, o charset e o gerenciamento de conexões da aplicação. Exemplo do conteúdo padrão:
```json
{
"host": "localhost",
"name": "D:/Orius/Base/CAIAPONIA.FDB",
"port": 3050,
"user": "SYSDBA",
"password": "",
"charset": "UTF8",
"pool": {
"pre_ping": true,
"size": 5,
"max_overflow": 10
}
}
```
### Ajustes Necessários:
* **host**: Endereço do servidor do banco de dados.
* **name**: Caminho completo do arquivo `.FDB`.
* **port**: Porta do Firebird (padrão: `3050`).
* **user**: Usuário do banco de dados.
* **password**: Senha do usuário configurado.
--- ---
## 7. Modo Desenvolvimento ## 7. Iniciar a Aplicação
Execute a aplicação em ambiente local: Com o ambiente virtual **ativado**, execute o comando abaixo para iniciar a aplicação:
```bash ```bash
uvicorn main:app --reload uvicorn main:app --reload
``` ```
O modo `--reload` recarrega a aplicação automaticamente ao detectar alterações no código. > **Dica:**
> O parâmetro `--reload` reinicia automaticamente a aplicação sempre que houver alterações no código.
Acesse a documentação da API em: ---
``` ## 8. Testando a Aplicação
Após iniciar a aplicação, abra o navegador e acesse o seguinte endereço:
```http
http://localhost:8000/docs http://localhost:8000/docs
``` ```
Opcionalmente, para expor a aplicação na rede: Você deverá visualizar a interface do **Swagger**, onde estarão listados todos os endpoints disponíveis da API.
```bash > **Observação:**
uvicorn main:app --host 0.0.0.0 --port 8000 --reload > O Swagger permite testar os endpoints diretamente pelo navegador, sem necessidade de ferramentas externas como Postman ou Insomnia.
```
--- ---
## 8. Modo Produção ## Resumo dos Comandos
A execução em produção varia conforme o sistema operacional. | Etapa | Comando |
| ----------------------- | ------------------------------------------------------------------------------- |
--- | Clonar o projeto | `git clone https://git.oriustecnologia.com/OriusTecnologia/saas_api.git` |
| Criar ambiente virtual | `python -m venv venv` |
### **Windows (modo produção simulado)** | Ativar ambiente virtual | `venv\Scripts\activate` *(Windows)*<br>`source venv/bin/activate` *(Linux/Mac)* |
| Instalar dependências | `pip install -r requirements.txt` |
O **Gunicorn** não é compatível com Windows, pois depende do módulo `fcntl` exclusivo de sistemas Unix. | Iniciar a aplicação | `uvicorn main:app --reload` |
Portanto, em ambiente Windows, recomenda-se usar o **Uvicorn** diretamente com múltiplos *workers*:
```bash
uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4
```
> O parâmetro `--workers` define quantos processos simultâneos serão utilizados.
> Idealmente, use `(número_de_CPUs * 2) + 1`.
#### Alternativa compatível (Windows)
Instale e use o **Hypercorn**, uma alternativa semelhante ao Gunicorn:
```bash
pip install hypercorn
hypercorn main:app --workers 4 --bind 0.0.0.0:8000
```
---
### **Linux (modo produção real)**
Em ambientes Linux (ou Docker), utilize o **Gunicorn** com o **Uvicorn Worker** para obter o máximo desempenho.
#### Instalar Gunicorn (caso ainda não instalado)
```bash
pip install gunicorn uvicorn
```
#### Executar com múltiplos núcleos
```bash
gunicorn main:app \
-k uvicorn.workers.UvicornWorker \
--workers 4 \
--bind 0.0.0.0:8000 \
--timeout 120 \
--log-level info
```
#### Parâmetros principais
| Parâmetro | Função |
| ---------------------------------- | ---------------------------------------- |
| `-k uvicorn.workers.UvicornWorker` | Usa o Uvicorn como worker ASGI |
| `--workers 4` | Define o número de núcleos usados |
| `--bind 0.0.0.0:8000` | Expõe a aplicação em todas as interfaces |
| `--timeout 120` | Tempo limite de resposta (em segundos) |
| `--log-level info` | Define o nível de logs |
#### Dica de cálculo de workers
```
(número_de_CPUs * 2) + 1
```
Exemplo: servidor com 2 CPUs → `--workers 5`
---
### **Execução em segundo plano (Linux)**
Para rodar a aplicação continuamente:
```bash
nohup gunicorn main:app -k uvicorn.workers.UvicornWorker --workers 4 --bind 0.0.0.0:8000 &
```
Verifique se está rodando:
```bash
ps aux | grep gunicorn
```
---
## 9. Logs e Monitoramento
É possível direcionar os logs de acesso e erro para arquivos dedicados:
```bash
gunicorn main:app \
-k uvicorn.workers.UvicornWorker \
--workers 4 \
--bind 0.0.0.0:8000 \
--access-logfile logs/access.log \
--error-logfile logs/error.log
```
---
## 10. Estrutura Recomendada de Deploy
```
/app
├── main.py
├── api/
├── packages/
├── requirements.txt
├── logs/
│ ├── access.log
│ └── error.log
└── systemd/
└── saas_api.service
```
---
## 11. Resumo dos Comandos
| Etapa | Comando |
| --------------------------- | ------------------------------------------------------------------------------------ |
| Clonar projeto | `git clone https://git.oriustecnologia.com/OriusTecnologia/saas_api.git` |
| Criar venv | `python -m venv venv` |
| Ativar venv | `venv\Scripts\activate` *(Windows)*<br>`source venv/bin/activate` *(Linux/Mac)* |
| Instalar dependências | `pip install -r requirements.txt` |
| Rodar em desenvolvimento | `uvicorn main:app --reload` |
| Rodar em produção (Windows) | `uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4` |
| Rodar em produção (Linux) | `gunicorn main:app -k uvicorn.workers.UvicornWorker --workers 4 --bind 0.0.0.0:8000` |
| Alternativa (Windows) | `hypercorn main:app --workers 4 --bind 0.0.0.0:8000` |
---
## 12. Recomendações Finais
* Em **Windows**, use Uvicorn ou Hypercorn apenas para testes e ambientes locais.
* Para **produção real**, use **Linux** com Gunicorn + Uvicorn Worker, idealmente em container **Docker**.
* Monitore o consumo de CPU/RAM e ajuste o número de *workers* conforme o ambiente.
* Automatize o serviço em produção via **systemd** (ex: `/etc/systemd/system/saas_api.service`) para iniciar junto com o servidor.
---

View file

@ -1,8 +1,7 @@
from abc import abstractmethod from abc import ABC, abstractmethod
class BaseAction: class BaseAction:
""" """
Classe abstrata base para todos as actions do sistema. Classe abstrata base para todos as actions do sistema.
Obriga implementação de um método execute(). Obriga implementação de um método execute().

View file

@ -1,114 +1,49 @@
# Importa tipos utilitários para anotações estáticas e sobrecarga de assinaturas. from typing import Literal, Optional
from typing import Any, Mapping, List, Optional, Literal, Union, overload
# Função `text` para construir SQL parametrizado de forma segura.
from sqlalchemy import text from sqlalchemy import text
# Tipo de retorno de execução bruta de SQL (cursor/result set) do SQLAlchemy.
from sqlalchemy.engine import CursorResult
# Exceção base do SQLAlchemy para capturar erros de banco.
from sqlalchemy.exc import SQLAlchemyError from sqlalchemy.exc import SQLAlchemyError
# Provedor do engine de conexão com o Firebird, centralizado no projeto.
from database.firebird import Firebird from database.firebird import Firebird
# Define a classe base de repositórios, concentrando operações comuns de acesso a dados.
class BaseRepository: class BaseRepository:
# Sobrecarga 1: quando `fetch="all"`, o retorno é uma lista de mapeamentos (coluna->valor). def query(self, sql: str, params: Optional[dict] = None):
@overload """Executa uma consulta SQL e retorna o resultado como objeto ResultProxy"""
def _execute(
self, sql: str, params: Optional[dict[str, Any]], fetch: Literal["all"]
) -> List[Mapping[str, Any]]: ...
# Sobrecarga 2: quando `fetch="one"`, o retorno é um único mapeamento ou None.
@overload
def _execute(
self, sql: str, params: Optional[dict[str, Any]], fetch: Literal["one"]
) -> Optional[Mapping[str, Any]]: ...
# Sobrecarga 3: quando `fetch="none"`, não há retorno (operações DML sem leitura).
@overload
def _execute(
self, sql: str, params: Optional[dict[str, Any]], fetch: Literal["none"]
) -> None: ...
# Sobrecarga 4: quando `fetch="result"`, retorna o objeto `CursorResult` bruto do SQLAlchemy.
@overload
def _execute(
self, sql: str, params: Optional[dict[str, Any]], fetch: Literal["result"]
) -> CursorResult[Any]: ...
# Implementação concreta que atende às quatro sobrecargas por meio de um retorno em união.
def _execute(
self,
sql: str, # Comando SQL (SELECT/INSERT/UPDATE/DELETE) em texto.
params: Optional[dict[str, Any]] = None, # Parâmetros nomeados para o SQL.
fetch: Literal[
"all", "one", "none", "result"
] = "result", # Modo de leitura/retorno.
) -> Union[
List[Mapping[str, Any]], # Retorno quando `fetch="all"`.
Optional[Mapping[str, Any]], # Retorno quando `fetch="one"`.
None, # Retorno quando `fetch="none"`.
CursorResult[Any], # Retorno quando `fetch="result"`.
]:
"""Método interno que executa o SQL conforme o modo de fetch especificado."""
# Obtém o engine de conexão com o Firebird a partir do provider central.
engine = Firebird.get_engine()
try:
# Inicia um contexto transacional; `begin()` garante commit/rollback automáticos.
with engine.begin() as conn:
# Executa o SQL com parâmetros (usa dict vazio quando `params` é None).
result = conn.execute(text(sql), params or {})
# Quando for solicitado "all", converte o resultado em lista de mapeamentos (coluna->valor).
if fetch == "all":
# retorna Sequence[RowMapping], compatível com List[Mapping[str, Any]]
return list(result.mappings().all())
# Quando for solicitado "one", retorna apenas o primeiro registro (ou None).
elif fetch == "one":
return result.mappings().first()
# Quando for solicitado "none", não retorna dados (apenas executa o comando).
elif fetch == "none":
return None
# Caso padrão: retorna o objeto Result bruto para manipulações específicas.
return result
except SQLAlchemyError as e:
# Log simples para facilitar diagnóstico em ambiente de desenvolvimento/produção.
print(f"[ERRO SQL]: {e}")
# Propaga a exceção para camadas superiores tratarem (ex.: FastAPI Exception Handler).
raise
# Executa uma consulta e retorna o objeto `CursorResult` bruto (uso avançado ou stream).
def query(
self, sql: str, params: Optional[dict[str, Any]] = None
) -> CursorResult[Any]:
"""Executa uma consulta SQL e retorna o resultado como objeto ResultProxy."""
return self._execute(sql, params, fetch="result") return self._execute(sql, params, fetch="result")
# Executa uma consulta e retorna todos os registros como lista de mapeamentos. def fetch_all(self, sql: str, params: Optional[dict] = None):
def fetch_all( """Executa uma consulta SQL e retorna todos os registros com mapeamento de colunas"""
self, sql: str, params: Optional[dict[str, Any]] = None
) -> List[Mapping[str, Any]]:
"""Executa uma consulta SQL e retorna todos os registros com mapeamento de colunas."""
return self._execute(sql, params, fetch="all") return self._execute(sql, params, fetch="all")
# Executa uma consulta e retorna apenas o primeiro registro ou None. def fetch_one(self, sql: str, params: Optional[dict] = None):
def fetch_one( """Executa uma consulta SQL e retorna o primeiro registro com mapeamento de colunas"""
self, sql: str, params: Optional[dict[str, Any]] = None
) -> Optional[Mapping[str, Any]]:
"""Executa uma consulta SQL e retorna o primeiro registro com mapeamento de colunas."""
return self._execute(sql, params, fetch="one") return self._execute(sql, params, fetch="one")
# Executa comandos DML (INSERT/UPDATE/DELETE) sem retorno de dados. def run(self, sql: str, params: Optional[dict] = None):
def run(self, sql: str, params: Optional[dict[str, Any]] = None) -> None: """Executa um SQL sem retorno de dados (ex: INSERT, UPDATE, DELETE)"""
"""Executa um SQL sem retorno de dados (ex: INSERT, UPDATE, DELETE).""" return self._execute(sql, params, fetch="none")
self._execute(sql, params, fetch="none")
# Executa comandos com cláusula RETURNING e devolve o registro retornado (ou None). def run_and_return(self, sql: str, params: Optional[dict] = None):
def run_and_return( """Executa SQL com RETURNING e retorna o primeiro registro como dict"""
self, sql: str, params: Optional[dict[str, Any]] = None
) -> Optional[Mapping[str, Any]]:
"""Executa SQL com RETURNING e retorna o primeiro registro como dict."""
return self._execute(sql, params, fetch="one") return self._execute(sql, params, fetch="one")
def _execute(
self,
sql: str,
params: Optional[dict] = None,
fetch: Literal["all", "one", "result", "none"] = "result",
):
engine = Firebird.get_engine()
try:
with engine.begin() as conn:
result = conn.execute(text(sql), params or {})
if fetch == "all":
return result.mappings().all()
elif fetch == "one":
return result.mappings().first()
elif fetch == "none":
return None
return result # Result object
except SQLAlchemyError as e:
print(f"[ERRO SQL]: {e}")
raise

View file

@ -1,31 +0,0 @@
import json
from types import SimpleNamespace
def dict_to_namespace(d):
"""
Converte dict (ou string JSON) recursivamente em SimpleNamespace.
"""
# Caso venha uma string JSON
if isinstance(d, str):
try:
# tenta fazer parse do JSON interno
parsed = json.loads(d)
# se for mesmo JSON, converte recursivamente
return dict_to_namespace(parsed)
except (json.JSONDecodeError, TypeError):
# não era JSON, retorna string normal
return d
# Caso seja um dicionário
if isinstance(d, dict):
return SimpleNamespace(**{k: dict_to_namespace(v) for k, v in d.items()})
# Caso seja lista/tupla
if isinstance(d, list):
return [dict_to_namespace(i) for i in d]
if isinstance(d, tuple):
return tuple(dict_to_namespace(i) for i in d)
# Caso base (valor simples)
return d

View file

@ -1,82 +0,0 @@
from typing import Any, Mapping, Iterable
class DictToObj:
"""
Converte dicts (aninhados) em objetos com acesso por ponto.
- d["x"] -> o.x
- Listas/tuplas são convertidas recursivamente.
- Mantém método parse() para voltar ao dict original.
"""
__slots__ = ("__data__",)
def __init__(self, data: Mapping[str, Any] | None = None):
object.__setattr__(self, "__data__", {})
if data:
for k, v in data.items():
self.__data__[k] = self._convert(v)
# ===== Conversões =====
@classmethod
def _convert(cls, value: Any) -> Any:
if isinstance(value, Mapping):
return cls(value)
if isinstance(value, list):
return [cls._convert(v) for v in value]
if isinstance(value, tuple):
return tuple(cls._convert(v) for v in value)
return value
def parse(self) -> dict[str, Any]:
def back(v: Any) -> Any:
if isinstance(v, DictToObj):
return v.parse()
if isinstance(v, list):
return [back(i) for i in v]
if isinstance(v, tuple):
return tuple(back(i) for i in v)
return v
return {k: back(v) for k, v in self.__data__.items()}
# ===== Acesso por ponto / item =====
def __getattr__(self, name: str) -> Any:
try:
return self.__data__[name]
except KeyError as e:
raise AttributeError(name) from e
def __setattr__(self, name: str, value: Any) -> None:
# protege o atributo interno
if name == "__data__":
object.__setattr__(self, name, value)
else:
self.__data__[name] = self._convert(value)
def __getitem__(self, key: str) -> Any:
return self.__data__[key]
def __setitem__(self, key: str, value: Any) -> None:
self.__data__[key] = self._convert(value)
def __contains__(self, key: str) -> bool:
return key in self.__data__
def keys(self) -> Iterable[str]:
return self.__data__.keys()
def items(self) -> Iterable[tuple[str, Any]]:
return self.__data__.items()
def values(self) -> Iterable[Any]:
return self.__data__.values()
def __iter__(self):
return iter(self.__data__)
def __len__(self) -> int:
return len(self.__data__)
def __repr__(self) -> str:
return f"DictToObj({self.__data__!r})"

View file

@ -1,51 +0,0 @@
from pydantic import BaseModel
from decimal import Decimal
def generate_insert_sql(table_name: str, data: BaseModel | dict) -> str:
"""
Gera um comando SQL INSERT seguro para Firebird.
- Aceita BaseModel (Pydantic) ou dict.
- Ignora *_ID None.
- Mantém colunas em MAIÚSCULAS e parâmetros em minúsculas (para bind funcionar).
"""
# Converte o model em dict limpo
if isinstance(data, BaseModel):
data_dict = data.model_dump(exclude_unset=True)
elif isinstance(data, dict):
data_dict = data
else:
raise TypeError("O parâmetro 'data' deve ser um BaseModel ou dict.")
columns = []
params = []
returning_fields = []
for key, value in data_dict.items():
column_name = key.upper()
# Converte Decimal → float
if isinstance(value, Decimal):
data_dict[key] = float(value)
# Campos válidos para retorno
if value is not None:
returning_fields.append(column_name)
# Coluna em maiúsculo, parâmetro em minúsculo
columns.append(column_name)
params.append(f":{key}")
columns_str = ", ".join(columns)
params_str = ", ".join(params)
returning_str = ", ".join(returning_fields) if returning_fields else "*"
sql = (
f"INSERT INTO {table_name} (\n"
f" {columns_str}\n"
f") VALUES (\n"
f" {params_str}\n"
f") RETURNING {returning_str};"
)
return sql

View file

@ -1,5 +0,0 @@
from fastapi import Request
async def get_url_params(request: Request):
return dict(request.query_params)

View file

@ -1,37 +0,0 @@
import json
from pathlib import Path
from typing import Any, Union
class JsonToDict:
"""
Converte conteúdo JSON (string, bytes ou arquivo) em dicionário Python.
"""
@staticmethod
def parse(data: Union[str, bytes, Path]) -> dict[str, Any]:
"""
Recebe uma string JSON, bytes ou caminho de arquivo .json
e retorna um dicionário Python.
"""
try:
# Caso seja um caminho de arquivo
if isinstance(data, Path):
with open(data, "r", encoding="utf-8") as file:
return json.load(file)
# Caso seja conteúdo JSON (str ou bytes)
if isinstance(data, bytes):
data = data.decode("utf-8")
# Garante que é string JSON
if isinstance(data, str):
return json.loads(data)
raise TypeError("Tipo de entrada inválido. Use str, bytes ou Path.")
except json.JSONDecodeError as e:
raise ValueError(f"Erro ao decodificar JSON: {e}")
except Exception as e:
raise ValueError(f"Erro ao converter JSON para dict: {e}")

View file

@ -1,31 +0,0 @@
import time
class Microtime:
"""
Utilitário para manipulação de tempo com alta precisão (microssegundos).
"""
@staticmethod
def get() -> float:
"""
Retorna o timestamp Unix atual com precisão de microssegundos.
Equivalente ao microtime(true) do PHP.
"""
return time.time()
@staticmethod
def as_int() -> int:
"""
Retorna o tempo atual puramente em microssegundos (Inteiro).
Útil para gerar IDs únicos ou ordenação precisa.
"""
# Pega em nanosegundos e converte para microssegundos
return time.time_ns() // 1000
@staticmethod
def diff(start_time: float) -> float:
"""
Calcula a diferença (duração) em segundos com precisão.
"""
return time.time() - start_time

View file

@ -1,30 +0,0 @@
from typing import Tuple, Dict, Any
from pydantic import BaseModel
def prepare_update_data(
schema: BaseModel, exclude_fields: list[str] = None, id_field: str = "id"
) -> Tuple[Dict[str, Any], str]:
"""
Gera dinamicamente os dados e SQL para update com base em um schema Pydantic.
Args:
schema: Instância do schema (ex: t_pessoa_save_schema)
exclude_fields: Lista de campos a serem ignorados no SET (ex: ['pessoa_id'])
id_field: Nome do campo identificador primário (ex: 'pessoa_id')
Returns:
Tuple contendo:
- data_dict: dicionário com apenas campos preenchidos (sem unset)
- update_sql: string SQL do tipo "campo1 = :campo1, campo2 = :campo2"
"""
exclude_fields = exclude_fields or [id_field]
# Cria o dicionário apenas com os campos enviados
data_dict = schema.model_dump(exclude_unset=True)
# Monta lista dinâmica de campos para o SET
update_fields = [f"{k} = :{k}" for k in data_dict.keys() if k not in exclude_fields]
update_sql = ", ".join(update_fields)
return data_dict, update_sql

View file

@ -1,128 +0,0 @@
# Importa a biblioteca nativa 'zlib' usada para compressão/descompressão de dados binários.
import zlib
# Importa a função 'rtf_to_text' da biblioteca 'striprtf',
# responsável por converter documentos RTF em texto plano legível.
from striprtf.striprtf import rtf_to_text
# Define uma classe utilitária chamada 'Text', contendo apenas métodos estáticos.
# Essa abordagem permite o uso direto sem necessidade de instanciar a classe.
class Text:
@staticmethod
def decompress(vf_string):
"""
Descomprime e decodifica texto de origem WPTools/Firebird.
Versão segura contra BLOB inválido (Firebird).
"""
if vf_string is None:
return ""
# ===============================
# 1) Tentativa segura de leitura
# ===============================
if hasattr(vf_string, "read"):
try:
vf_string = vf_string.read()
except Exception:
# BLOB inválido, conexão fechada ou handle perdido
return ""
if not vf_string:
return ""
# ===============================
# 2) Garantir bytes
# ===============================
if isinstance(vf_string, str):
vf_bytes = vf_string.encode("latin1", errors="ignore")
else:
try:
vf_bytes = bytes(vf_string)
except Exception:
return ""
# ===============================
# 3) Detectar zlib
# ===============================
is_zlib = (
len(vf_bytes) > 2
and vf_bytes[0] == 0x78
and vf_bytes[1] in (0x01, 0x9C, 0xDA)
)
# ===============================
# 4) Descompactar se necessário
# ===============================
if is_zlib:
try:
return zlib.decompress(vf_bytes).decode("iso-8859-1", errors="ignore")
except Exception:
# Se falhar, tenta como texto puro
pass
# ===============================
# 5) Texto puro
# ===============================
try:
return vf_bytes.decode("iso-8859-1", errors="ignore")
except Exception:
return ""
@staticmethod
def compress(text, *, encoding: str = "iso-8859-1"):
"""
Comprime texto/dados com zlib SEM Base64.
Parâmetros:
text: str | bytes | stream (com .read())
encoding: encoding usado quando 'text' for str (padrão: ISO-8859-1)
Retorno:
- bytes comprimidos (zlib)
Observações:
- Ideal para armazenamento direto em BLOB (Firebird, PostgreSQL, etc.)
- Evita overhead e custo do Base64
- Totalmente compatível com 'decompress'
"""
if text is None or text == "":
return b""
# Se for stream (ex.: BLOB do Firebird)
if hasattr(text, "read"):
raw = text.read()
else:
raw = text
# Garante bytes
if isinstance(raw, str):
raw_bytes = raw.encode(encoding, errors="ignore")
else:
raw_bytes = bytes(raw)
# Comprime com zlib e retorna bytes
return zlib.compress(raw_bytes)
@staticmethod
def to_text(raw_text: str) -> str:
"""
Converte o conteúdo RTF em texto simples e retorna como string.
Finalidade:
- Detectar automaticamente se o conteúdo está em formato RTF.
- Converter para texto plano usando a função 'rtf_to_text'.
- Retornar uma string limpa e pronta para uso.
"""
if not raw_text:
return ""
# Detecta cabeçalho RTF
if raw_text.strip().startswith("{\\rtf"):
try:
return rtf_to_text(raw_text).strip()
except Exception:
return raw_text
return raw_text.strip()

View file

@ -1,6 +1,7 @@
import importlib import importlib
from typing import Any, Optional, Type
from actions.config.config import Config from actions.config.config import Config
from typing import Optional, Any, Type
class DynamicImport: class DynamicImport:

View file

@ -1,64 +0,0 @@
import importlib
import re
from functools import lru_cache
from typing import Type, TypeVar, Optional
from actions.config.config import Config
from actions.env.env_config_loader import EnvConfigLoader
# Genérico para garantir que o retorno respeite o Protocolo
T = TypeVar("T")
class ServiceFactory:
def __init__(self, package: str, table: str):
# Instancia o loader com o prefixo correto
env = EnvConfigLoader(".env")
# Ex: "packages.v1"
self.base_root = "packages.v1"
self.package = package
self.table = table
# Carrega config apenas uma vez
self.app_config = Config.get("app.json")
# Define a UF da aplicação
self.current_state = env.ORIUS_CLIENT_STATE
@lru_cache(maxsize=32)
def make(self, class_name: str, interface: Type[T]) -> T:
"""
Instancia um serviço dinamicamente com comportamento de Autoload.
"""
# 1. Converte CamelCase para snake_case (Autoload style)
# Ex: TAtoIndexService -> t_ato_index_service
module_name = re.sub(r"(?<!^)(?=[A-Z])", "_", class_name).lower()
# 2. Monta o caminho completo
# Ex: packages.v1.servicos.balcao.services.t_ato.SP.t_ato_index_service
import_path = (
f"{self.base_root}.{self.package}.services."
f"{self.table}.{self.current_state}.{module_name}"
)
try:
# 3. Importação Dinâmica
module = importlib.import_module(import_path)
# 4. Pega a classe do módulo
clazz = getattr(module, class_name)
# 5. Retorna a INSTÂNCIA da classe (já com () )
# Se seus serviços precisam de argumentos no __init__, altere aqui.
return clazz()
except ImportError as e:
raise ImportError(
f"FATAL: Não foi possível carregar o serviço '{class_name}' para o estado '{self.current_state}'.\nCaminho tentado: {import_path}\nErro: {e}"
)
except AttributeError:
raise AttributeError(
f"FATAL: O arquivo '{module_name}.py' existe, mas a classe '{class_name}' não foi encontrada dentro dele."
)

View file

@ -1,36 +0,0 @@
from dotenv import dotenv_values, load_dotenv
import os
class EnvConfigLoader:
"""
Classe para carregar todas as variáveis de ambiente (.env)
e permitir acesso via atributo (settings.VAR).
"""
def __init__(self, env_file: str = ".env"):
# Garante que o .env será carregado no ambiente do sistema
load_dotenv(env_file)
# Lê todas as variáveis (do arquivo + sistema)
self._values = {
**dotenv_values(env_file), # Conteúdo do .env
**os.environ, # Variáveis já existentes no ambiente
}
def __getattr__(self, name: str):
"""Permite acessar como settings.VAR"""
# Normaliza o nome para maiúsculo
key = name.upper()
if key in self._values:
return self._values[key]
raise AttributeError(f"A variável '{name}' não existe no .env")
def __repr__(self):
"""Exibe todas as variáveis carregadas"""
return f"<Settings {self._values}>"
def all(self) -> dict:
"""Retorna todas as variáveis como dicionário"""
return dict(self._values)

View file

@ -4,14 +4,14 @@ import os
class File: class File:
def create(self, data, caminho_arquivo='storage/temp.json'): def create(self, data, caminho_arquivo="storage/temp.json"):
try: try:
# Garante que a pasta existe # Garante que a pasta existe
os.makedirs(os.path.dirname(caminho_arquivo), exist_ok=True) os.makedirs(os.path.dirname(caminho_arquivo), exist_ok=True)
# Lê dados existentes (ou cria nova lista) # Lê dados existentes (ou cria nova lista)
if os.path.exists(caminho_arquivo): if os.path.exists(caminho_arquivo):
with open(caminho_arquivo, 'r', encoding='utf-8') as arquivo: with open(caminho_arquivo, "r", encoding="utf-8") as arquivo:
try: try:
dados_existentes = json.load(arquivo) dados_existentes = json.load(arquivo)
if not isinstance(dados_existentes, list): if not isinstance(dados_existentes, list):
@ -25,7 +25,7 @@ class File:
dados_existentes.append(data) dados_existentes.append(data)
# Salva novamente no arquivo com indentação # Salva novamente no arquivo com indentação
with open(caminho_arquivo, 'w', encoding='utf-8') as arquivo: with open(caminho_arquivo, "w", encoding="utf-8") as arquivo:
json.dump(dados_existentes, arquivo, indent=4, ensure_ascii=False) json.dump(dados_existentes, arquivo, indent=4, ensure_ascii=False)
except Exception as e: except Exception as e:

View file

@ -1,4 +1,5 @@
from datetime import datetime, timedelta from datetime import datetime, timedelta
from jose import jwt from jose import jwt
from pytz import timezone from pytz import timezone
@ -8,7 +9,6 @@ from actions.config.config import Config
class CreateToken(BaseAction): class CreateToken(BaseAction):
def __init__(self): def __init__(self):
# Busca as configurações da aplicação # Busca as configurações da aplicação
self.config = Config.get("app.json") self.config = Config.get("app.json")

View file

@ -1,7 +1,8 @@
from fastapi import Depends, HTTPException, status from fastapi import Depends, HTTPException, Request, status
from fastapi.security import OAuth2PasswordBearer from fastapi.security import OAuth2PasswordBearer
from actions.jwt.verify_token import VerifyToken # A classe que criamos anteriormente from actions.jwt.verify_token import \
VerifyToken # A classe que criamos anteriormente
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token") # Apenas requerido pelo FastAPI oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token") # Apenas requerido pelo FastAPI

View file

@ -1,4 +1,7 @@
from jose import jwt, JWTError, ExpiredSignatureError from datetime import datetime
from jose import ExpiredSignatureError, JWTError, jwt
from pytz import timezone
from actions.config.config import Config from actions.config.config import Config
@ -6,15 +9,13 @@ from actions.config.config import Config
class VerifyToken: class VerifyToken:
def __init__(self): def __init__(self):
# Carrega configurações # Carrega configurações
self.config = Config.get('app.json') self.config = Config.get("app.json")
def execute(self, token: str, expected_type: str = 'access-token') -> dict: def execute(self, token: str, expected_type: str = "access-token") -> dict:
try: try:
# Decodifica o token # Decodifica o token
payload = jwt.decode( payload = jwt.decode(
token, token, self.config.jwt.token, algorithms=[self.config.jwt.algorithm]
self.config.jwt.token,
algorithms=[self.config.jwt.algorithm]
) )
# Valida expiração # Valida expiração
@ -31,25 +32,16 @@ class VerifyToken:
if "data" not in payload: if "data" not in payload:
raise ValueError("Token malformado: campo 'data' ausente.") raise ValueError("Token malformado: campo 'data' ausente.")
return { return {"status": "valid", "payload": payload}
"status": "valid",
"payload": payload
}
except ExpiredSignatureError: except ExpiredSignatureError:
return { return {"status": "expired", "message": "O token expirou."}
"status": "expired",
"message": "O token expirou."
}
except JWTError as e: except JWTError as e:
return { return {"status": "invalid", "message": f"Token inválido: {str(e)}"}
"status": "invalid",
"message": f"Token inválido: {str(e)}"
}
except Exception as e: except Exception as e:
return { return {
"status": "error", "status": "error",
"message": f"Erro na validação do token: {str(e)}" "message": f"Erro na validação do token: {str(e)}",
} }

View file

@ -4,14 +4,14 @@ import os
class Log: class Log:
def register(self, data, caminho_arquivo='storage/temp.json'): def register(self, data, caminho_arquivo="storage/temp.json"):
try: try:
# Garante que a pasta existe # Garante que a pasta existe
os.makedirs(os.path.dirname(caminho_arquivo), exist_ok=True) os.makedirs(os.path.dirname(caminho_arquivo), exist_ok=True)
# Lê dados existentes (ou cria nova lista) # Lê dados existentes (ou cria nova lista)
if os.path.exists(caminho_arquivo): if os.path.exists(caminho_arquivo):
with open(caminho_arquivo, 'r', encoding='utf-8') as arquivo: with open(caminho_arquivo, "r", encoding="utf-8") as arquivo:
try: try:
dados_existentes = json.load(arquivo) dados_existentes = json.load(arquivo)
if not isinstance(dados_existentes, list): if not isinstance(dados_existentes, list):
@ -25,7 +25,7 @@ class Log:
dados_existentes.append(data) dados_existentes.append(data)
# Salva novamente no arquivo com indentação # Salva novamente no arquivo com indentação
with open(caminho_arquivo, 'w', encoding='utf-8') as arquivo: with open(caminho_arquivo, "w", encoding="utf-8") as arquivo:
json.dump(dados_existentes, arquivo, indent=4, ensure_ascii=False) json.dump(dados_existentes, arquivo, indent=4, ensure_ascii=False)
except Exception as e: except Exception as e:

View file

@ -6,7 +6,7 @@ from passlib.context import CryptContext
# Cria uma instância do contexto de criptografia # Cria uma instância do contexto de criptografia
# O esquema usado é 'bcrypt', que é seguro e amplamente aceito # O esquema usado é 'bcrypt', que é seguro e amplamente aceito
# O parâmetro 'deprecated="auto"' marca versões antigas como inseguras, se aplicável # O parâmetro 'deprecated="auto"' marca versões antigas como inseguras, se aplicável
CRYPTO = CryptContext(schemes=['bcrypt'], deprecated='auto') CRYPTO = CryptContext(schemes=["bcrypt"], deprecated="auto")
class Security: class Security:
@ -19,7 +19,6 @@ class Security:
""" """
return CRYPTO.identify(senha) return CRYPTO.identify(senha)
# Verifica se uma senha fornecida corresponde ao hash armazenado # Verifica se uma senha fornecida corresponde ao hash armazenado
def verify_senha_api(plain_senha_api: str, hashed_senha_api: str) -> bool: def verify_senha_api(plain_senha_api: str, hashed_senha_api: str) -> bool:
""" """
@ -31,7 +30,6 @@ class Security:
""" """
return CRYPTO.verify(plain_senha_api, hashed_senha_api) return CRYPTO.verify(plain_senha_api, hashed_senha_api)
# Gera o hash de uma senha fornecida # Gera o hash de uma senha fornecida
def hash_senha_api(plain_senha_api: str) -> str: def hash_senha_api(plain_senha_api: str) -> str:
""" """

View file

@ -1,11 +0,0 @@
from fastapi import Request, HTTPException, status
def get_session_user(request: Request) -> dict:
user = request.session.get("user")
if not user:
# ajuste conforme sua regra (pode só retornar None)
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED, detail="Sessão inválida"
)
return user

View file

@ -1,16 +0,0 @@
# services/session_service.py
from fastapi import Request
class SessionService:
def __init__(self, request: Request):
self._session = request.session
def set(self, k, v):
self._session[k] = v
def get(self, k, d=None):
return self._session.get(k, d)
def clear(self):
self._session.clear()

View file

@ -1,4 +1,5 @@
# handlers.py # handlers.py
import json
import traceback import traceback
from fastapi import Request from fastapi import Request
@ -6,8 +7,8 @@ from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse from fastapi.responses import JSONResponse
from starlette.exceptions import HTTPException as StarletteHTTPException from starlette.exceptions import HTTPException as StarletteHTTPException
from actions.system.exceptions import BusinessRuleException
from actions.log.log import Log from actions.log.log import Log
from actions.system.exceptions import BusinessRuleException
def register_exception_handlers(app): def register_exception_handlers(app):

View file

@ -7,7 +7,7 @@ class CNPJ:
def validate(data: str) -> bool: def validate(data: str) -> bool:
# Remove caracteres não numéricos # Remove caracteres não numéricos
data = re.sub(r'\D', '', data) data = re.sub(r"\D", "", data)
# Verifica se tem 14 dígitos # Verifica se tem 14 dígitos
if len(data) != 14: if len(data) != 14:
@ -21,7 +21,7 @@ class CNPJ:
def calcular_digito(data, peso): def calcular_digito(data, peso):
soma = sum(int(a) * b for a, b in zip(data, peso)) soma = sum(int(a) * b for a, b in zip(data, peso))
resto = soma % 11 resto = soma % 11
return '0' if resto < 2 else str(11 - resto) return "0" if resto < 2 else str(11 - resto)
# Primeiro dígito verificador # Primeiro dígito verificador
peso1 = [5, 4, 3, 2, 9, 8, 7, 6, 5, 4, 3, 2] peso1 = [5, 4, 3, 2, 9, 8, 7, 6, 5, 4, 3, 2]

View file

@ -6,7 +6,7 @@ class CPF:
@staticmethod @staticmethod
def is_valid_cpf(data: str) -> bool: def is_valid_cpf(data: str) -> bool:
# Remove caracteres não numéricos # Remove caracteres não numéricos
data = re.sub(r'\D', '', data) data = re.sub(r"\D", "", data)
# Verifica se tem 11 dígitos # Verifica se tem 11 dígitos
if len(data) != 11: if len(data) != 11:
@ -20,7 +20,7 @@ class CPF:
def calcular_digito(digitos, peso): def calcular_digito(digitos, peso):
soma = sum(int(a) * b for a, b in zip(digitos, peso)) soma = sum(int(a) * b for a, b in zip(digitos, peso))
resto = soma % 11 resto = soma % 11
return '0' if resto < 2 else str(11 - resto) return "0" if resto < 2 else str(11 - resto)
# Primeiro dígito verificador # Primeiro dígito verificador
peso1 = range(10, 1, -1) peso1 = range(10, 1, -1)

View file

@ -1,4 +1,3 @@
class Phone: class Phone:
@staticmethod @staticmethod

View file

@ -7,7 +7,7 @@ class Text:
# Remove as mascaras de números # Remove as mascaras de números
@staticmethod @staticmethod
def just_numbers(data: str) -> str: def just_numbers(data: str) -> str:
""" Mantêm apenas os numeros """ """Mantêm apenas os numeros"""
data = re.sub(r"[^\d]", "", data) data = re.sub(r"[^\d]", "", data)
return data return data
@ -17,7 +17,6 @@ class Text:
"""Check if email has a valid structure""" """Check if email has a valid structure"""
return bool(re.match(r"^[\w\.-]+@[\w\.-]+\.\w+$", email)) return bool(re.match(r"^[\w\.-]+@[\w\.-]+\.\w+$", email))
""" """
Sanitiza entradas de texto contra XSS e SQL Injection básicos. Sanitiza entradas de texto contra XSS e SQL Injection básicos.
- Remove espaços extras - Remove espaços extras
@ -25,6 +24,7 @@ class Text:
- Remove padrões suspeitos de XSS e SQL Injection - Remove padrões suspeitos de XSS e SQL Injection
- Normaliza múltiplos espaços em um - Normaliza múltiplos espaços em um
""" """
@staticmethod @staticmethod
def sanitize_input(data: str) -> str: def sanitize_input(data: str) -> str:
@ -41,19 +41,32 @@ class Text:
data = re.sub(r"\s+", " ", data) data = re.sub(r"\s+", " ", data)
# 4) Remove tags <script> (com atributos) # 4) Remove tags <script> (com atributos)
data = re.sub(r'<\s*script[^>]*>', '', data, flags=re.IGNORECASE) data = re.sub(r"<\s*script[^>]*>", "", data, flags=re.IGNORECASE)
data = re.sub(r'<\s*/\s*script\s*>', '', data, flags=re.IGNORECASE) data = re.sub(r"<\s*/\s*script\s*>", "", data, flags=re.IGNORECASE)
# 5) Remove javascript: de links # 5) Remove javascript: de links
data = re.sub(r'javascript\s*:', '', data, flags=re.IGNORECASE) data = re.sub(r"javascript\s*:", "", data, flags=re.IGNORECASE)
# 6) Remove palavras-chave SQL Injection comuns # 6) Remove palavras-chave SQL Injection comuns
blacklist = [ blacklist = [
"--", ";", "/*", "*/", "@@", "--",
"char(", "nchar(", "varchar(", ";",
"alter", "drop", "exec", "insert", "/*",
"delete", "update", "union", "select", "*/",
"from", "where" "@@",
"char(",
"nchar(",
"varchar(",
"alter",
"drop",
"exec",
"insert",
"delete",
"update",
"union",
"select",
"from",
"where",
] ]
for word in blacklist: for word in blacklist:
# Verificar se 'word' é uma string não vazia e válida para a regex # Verificar se 'word' é uma string não vazia e válida para a regex

View file

@ -1,14 +0,0 @@
from decimal import ROUND_HALF_UP, Decimal
class Values:
@staticmethod
def percent(value: Decimal, percent: Decimal) -> Decimal:
return value / percent
@staticmethod
def money(scale: str, value: Decimal) -> Decimal:
return value.quantize(scale, rounding=ROUND_HALF_UP)

View file

@ -1,186 +0,0 @@
{
"folders": [
{
"path": "D:/IIS/Orius/api"
}
],
"settings": {
// === ⚙️ GERAL ===
"editor.formatOnSave": true,
"editor.codeActionsOnSave": {
"source.fixAll": "explicit",
"source.organizeImports": "explicit"
},
"editor.formatOnPaste": false,
"editor.formatOnType": false,
"editor.minimap.enabled": false,
"files.trimTrailingWhitespace": true,
"files.autoSave": "onFocusChange",
"telemetry.telemetryLevel": "off",
"update.mode": "manual",
"workbench.startupEditor": "none",
"workbench.editor.enablePreview": false,
// === ⚡ PERFORMANCE ===
"files.watcherExclude": {
"**/__pycache__/**": true,
"**/.mypy_cache/**": true,
"**/.pytest_cache/**": true,
"**/.venv/**": true,
"**/venv/**": true,
"**/.git/**": true
},
"search.exclude": {
"**/__pycache__": true,
"**/.git": true,
"**/.mypy_cache": true,
"**/.pytest_cache": true
},
// === 🐍 PYTHON ===
"python.defaultInterpreterPath": "D:/IIS/Orius/api/venv/Scripts/python.exe",
"python.languageServer": "Pylance",
"python.analysis.autoImportCompletions": true,
"python.analysis.indexing": true,
"python.analysis.typeCheckingMode": "basic",
"python.analysis.useLibraryCodeForTypes": true,
"python.analysis.memory.keepLibraryAst": false,
// === 🧹 FORMATADOR ===
"python.formatting.provider": "black",
"python.formatting.blackArgs": [
"--line-length",
"100"
],
// === ✅ LINTING ===
"python.linting.enabled": true,
"python.linting.pylintEnabled": false,
"python.linting.flake8Enabled": true,
"python.linting.flake8Args": [
"--max-line-length=100"
],
// === 🧩 FORMATAÇÃO LOCAL ===
"[python]": {
"editor.defaultFormatter": "ms-python.black-formatter",
"editor.formatOnSave": true
},
// === 🧠 GIT ===
"git.enabled": true,
"git.autorefresh": true,
"git.confirmSync": false,
"git.fetchOnPull": true,
"git.countBadge": "tracked",
"git.pruneOnFetch": true,
"git.postCommitCommand": "sync",
"git.openDiffOnClick": true,
"git.decorations.enabled": true,
// === 🔍 GITLENS ===
"gitlens.advanced.messages": {
"suppressShowKeyBindingsNotice": true
},
"gitlens.codeLens.enabled": false,
"gitlens.currentLine.enabled": false,
"gitlens.hovers.enabled": true,
"gitlens.views.repositories.autoRefresh": true,
"gitlens.views.repositories.location": "scm",
"gitlens.defaultDateFormat": "DD/MM/YYYY HH:mm",
"gitlens.defaultDateShortFormat": "DD/MM HH:mm",
"gitlens.statusBar.enabled": true,
// === 💻 TERMINAIS INTEGRADOS ===
"terminal.integrated.profiles.windows": {
"FastAPI Dev": {
"path": "cmd.exe",
"args": [
"/k",
"cd D:\\IIS\\Orius\\api && venv\\Scripts\\activate && uvicorn main:app --reload --log-level debug"
]
},
"FastAPI Prod": {
"path": "cmd.exe",
"args": [
"/k",
"cd D:\\IIS\\Orius\\api && venv\\Scripts\\activate && uvicorn main:app --host 0.0.0.0 --port 8000"
]
},
"FastAPI Test": {
"path": "cmd.exe",
"args": [
"/k",
"cd D:\\IIS\\Orius\\api && venv\\Scripts\\activate && pytest -v"
]
},
"Python Shell": {
"path": "cmd.exe",
"args": [
"/k",
"cd D:\\IIS\\Orius\\api && venv\\Scripts\\activate"
]
},
"Git Bash": {
"path": "C:\\Program Files\\Git\\bin\\bash.exe"
}
},
"terminal.integrated.defaultProfile.windows": "FastAPI Dev",
"terminal.integrated.scrollback": 10000,
"terminal.integrated.enablePersistentSessions": false,
// === 🧪 TESTES ===
"python.testing.pytestEnabled": true,
"python.testing.unittestEnabled": false,
"python.testing.autoTestDiscoverOnSaveEnabled": true,
// === 🗂️ ARQUIVOS ===
"files.exclude": {
"**/.DS_Store": true,
"**/*.log": true
}
},
"launch": {
"version": "0.2.0",
"configurations": [
{
"name": "Debug FastAPI",
"type": "python",
"request": "launch",
"module": "uvicorn",
"args": [
"main:app",
"--reload",
"--host",
"127.0.0.1",
"--port",
"8000"
],
"jinja": true,
"justMyCode": true,
"cwd": "${workspaceFolder}",
"env": {
"PYTHONPATH": "${workspaceFolder}"
}
},
{
"name": "Pytest All",
"type": "python",
"request": "launch",
"program": "${workspaceFolder}/venv/Scripts/pytest.exe",
"args": [
"-v"
],
"console": "integratedTerminal"
}
]
},
"extensions": {
"recommendations": [
// === 🐍 PYTHON ===
"ms-python.python",
"ms-python.pylance",
"ms-python.black-formatter",
"ms-python.flake8",
"littlefoxteam.vscode-python-test-adapter",
// === 🔍 GIT ===
"eamodio.gitlens",
"mhutchie.git-graph",
"donjayamanne.githistory",
// === 🧰 UTILITÁRIOS ===
"formulahendry.code-runner",
"streetsidesoftware.code-spell-checker",
"tamasfe.even-better-toml"
]
}
}

View file

@ -1,13 +0,0 @@
{
"host": "localhost",
"name": "/database/BASE.FDB",
"port": 3050,
"user": "SYSDBA",
"password": "302b3c",
"charset": "UTF8",
"pool" : {
"pre_ping" : true,
"size" : 5,
"max_overflow" :10
}
}

View file

@ -1,7 +1,9 @@
from typing import Optional from typing import Optional
from sqlalchemy import create_engine from sqlalchemy import create_engine
from sqlalchemy.engine import Engine from sqlalchemy.engine import Engine
from actions.env.env_config_loader import EnvConfigLoader
from actions.config.config import Config
class Firebird: class Firebird:
@ -9,29 +11,28 @@ class Firebird:
@classmethod @classmethod
def get_engine(cls) -> Engine: def get_engine(cls) -> Engine:
# Obtem as configurações do arquivo JSON
# Instancia o loader com o prefixo correto database = Config.get("database/firebird.json")
env_database = EnvConfigLoader(".env")
# Cria a engine apenas uma vez # Cria a engine apenas uma vez
if cls._engine is None: if cls._engine is None:
# DSN para o SQLAlchemy usando firebird-driver # DSN para o SQLAlchemy usando firebird-driver
dsn = ( dsn = (
f"firebird+firebird://{env_database.ORIUS_API_FDB_USER}:" f"firebird+firebird://{database.user}:"
f"{env_database.ORIUS_API_FDB_PASSWORD}@" f"{database.password}@"
f"{env_database.ORIUS_API_FDB_HOST}:" f"{database.host}:"
f"{env_database.ORIUS_API_FDB_PORT}/" f"{database.port}/"
f"{env_database.ORIUS_API_FDB_NAME}" f"{database.name}"
) )
# Criação da engine SQLAlchemy # Criação da engine SQLAlchemy
cls._engine = create_engine( cls._engine = create_engine(
dsn, dsn,
connect_args={"charset": env_database.ORIUS_API_FDB_CHARSET}, connect_args={"charset": database.charset},
pool_pre_ping=bool(env_database.ORIUS_API_FDB_POOL_PRE_PING), pool_pre_ping=bool(database.pool.pre_ping),
pool_size=int(env_database.ORIUS_API_FDB_POOL_SIZE), pool_size=database.pool.size,
max_overflow=int(env_database.ORIUS_API_FDB_POOL_MAX_OVERFLOW), max_overflow=database.pool.max_overflow,
) )
return cls._engine return cls._engine

17
env.bat
View file

@ -1,17 +0,0 @@
@echo off
echo === Registrando variáveis de ambiente ORIUS_API ===
setx ORIUS_API_FDB_HOST "localhost"
setx ORIUS_API_FDB_NAME "S:\Bases\SANTARITA.FDB"
setx ORIUS_API_FDB_PORT "3050"
setx ORIUS_API_FDB_USER "SYSDBA"
setx ORIUS_API_FDB_PASSWORD "302b3c"
setx ORIUS_API_FDB_CHARSET "UTF8"
setx ORIUS_API_FDB_POOL_PRE_PING "true"
setx ORIUS_API_FDB_POOL_SIZE "5"
setx ORIUS_API_FDB_POOL_MAX_OVERFLOW "10"
echo.
echo Variáveis ORIUS_API registradas com sucesso!
echo Reinicie o terminal ou o computador para aplicar as alterações.
pause

View file

@ -1,10 +0,0 @@
from typing import Protocol, Any, runtime_checkable
@runtime_checkable
class ServiceProtocolsInterface(Protocol):
"""
Contrato que garante que todo serviço tenha um método execute.
"""
def execute(self, schema: Any) -> Any: ...

43
main.py
View file

@ -1,41 +1,27 @@
# Ajuste para garantir que o diretório base do projeto seja incluído no PYTHONPATH # Ajuste para garantir que o diretório base do projeto seja incluído no PYTHONPATH
import os import os
import platform
import sys import sys
from starlette.middleware.sessions import SessionMiddleware
# Adiciona o diretório atual (onde está o main.py) ao sys.path # Adiciona o diretório atual (onde está o main.py) ao sys.path
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))) sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
from firebird.driver import driver_config from pathlib import Path
# Importa a classe principal do FastAPI # Importa a classe principal do FastAPI
from fastapi import FastAPI, Request from fastapi import FastAPI, Request
from pathlib import Path
# Importa o middleware de CORS # Importa o middleware de CORS
from fastapi.middleware.cors import CORSMiddleware from fastapi.middleware.cors import CORSMiddleware
from fastapi.staticfiles import StaticFiles from fastapi.responses import Response
from starlette.middleware.base import BaseHTTPMiddleware
# Importa o roteador principal da API versão 1
from packages.v1.api import api_router
from packages.v1.system.service.startup_check_service import StartupCheckService
from actions.config.config import Config
# Importa as configurações globais da aplicação # Importa as configurações globais da aplicação
from actions.log.log import Log from actions.log.log import Log
from actions.config.config import Config
from actions.system.handlers import register_exception_handlers from actions.system.handlers import register_exception_handlers
# Importa o roteador principal da API versão 1
if platform.system() == "Windows": from packages.v1.api import api_router
from packages.v1.system.service.startup_check_service import \
# Ajuste o caminho abaixo para onde o Firebird está instalado: StartupCheckService
FB_DIR = r"C:\Program Files\Firebird\Firebird_4_0"
os.add_dll_directory(FB_DIR)
driver_config.fb_client_library.value = (
r"C:\Program Files\Firebird\Firebird_4_0\fbclient.dll"
)
config = Config.get("app.json") config = Config.get("app.json")
@ -48,23 +34,12 @@ register_exception_handlers(app)
# Adiciona o middleware de CORS # Adiciona o middleware de CORS
app.add_middleware( app.add_middleware(
CORSMiddleware, CORSMiddleware,
allow_origins=["*"], # Domínio do frontend allow_origins=["http://localhost:3000"], # Domínio do frontend
allow_credentials=True, allow_credentials=True,
allow_methods=["*"], allow_methods=["*"],
allow_headers=["*"], allow_headers=["*"],
) )
app.add_middleware(
SessionMiddleware,
secret_key="coloque-uma-secret-bem-grande-e-aleatoria",
session_cookie="sid",
same_site="lax",
https_only=True,
max_age=60 * 60 * 8,
)
app.mount(path="/temp", app=StaticFiles(directory="./storage/temp"), name="temp")
@app.on_event("startup") @app.on_event("startup")
async def on_startup(): async def on_startup():

View file

@ -1,11 +1,13 @@
from packages.v1.administrativo.repositories.c_caixa_item.c_caixa_item_delete import Delete
from packages.v1.administrativo.schemas.c_caixa_item_schema import CaixaItemSchema
from abstracts.action import BaseAction from abstracts.action import BaseAction
from packages.v1.administrativo.repositories.c_caixa_item.c_caixa_item_delete import \
Delete
from packages.v1.administrativo.schemas.c_caixa_item_schema import \
CaixaItemSchema
class DeleteAction(BaseAction): class DeleteAction(BaseAction):
def execute(self, caixa_item_schema : CaixaItemSchema): def execute(self, caixa_item_schema: CaixaItemSchema):
# Instânciamento de repoistório # Instânciamento de repoistório
delete = Delete() delete = Delete()

View file

@ -1,6 +1,8 @@
from abstracts.action import BaseAction
from packages.v1.administrativo.repositories.c_caixa_item.c_caixa_item_index import \ from packages.v1.administrativo.repositories.c_caixa_item.c_caixa_item_index import \
Index Index
from abstracts.action import BaseAction from packages.v1.administrativo.schemas.c_caixa_item_schema import \
CaixaItemSearchSchema
class IndexAction(BaseAction): class IndexAction(BaseAction):

View file

@ -1,12 +1,13 @@
from packages.v1.administrativo.repositories.c_caixa_item.c_caixa_item_save import Save from abstracts.action import BaseAction
from packages.v1.administrativo.repositories.c_caixa_item.c_caixa_item_save import \
Save
from packages.v1.administrativo.schemas.c_caixa_item_schema import \ from packages.v1.administrativo.schemas.c_caixa_item_schema import \
CaixaItemSchema CaixaItemSchema
from abstracts.action import BaseAction
class SaveAction(BaseAction): class SaveAction(BaseAction):
def execute(self, caixa_item_schema : CaixaItemSchema): def execute(self, caixa_item_schema: CaixaItemSchema):
# Instância o repositório desejado # Instância o repositório desejado
save = Save() save = Save()

View file

@ -1,11 +1,13 @@
from packages.v1.administrativo.repositories.c_caixa_item.c_caixa_item_show import Show
from packages.v1.administrativo.schemas.c_caixa_item_schema import CaixaItemSchema
from abstracts.action import BaseAction from abstracts.action import BaseAction
from packages.v1.administrativo.repositories.c_caixa_item.c_caixa_item_show import \
Show
from packages.v1.administrativo.schemas.c_caixa_item_schema import \
CaixaItemSchema
class ShowAction(BaseAction): class ShowAction(BaseAction):
def execute(self, caixa_item_schema : CaixaItemSchema): def execute(self, caixa_item_schema: CaixaItemSchema):
# Instânciamento do repositório # Instânciamento do repositório
show = Show() show = Show()

View file

@ -1,11 +1,13 @@
from packages.v1.administrativo.repositories.c_caixa_item.c_caixa_item_update_repository import UpdateRepository
from packages.v1.administrativo.schemas.c_caixa_item_schema import CaixaItemSchema
from abstracts.action import BaseAction from abstracts.action import BaseAction
from packages.v1.administrativo.repositories.c_caixa_item.c_caixa_item_update_repository import \
UpdateRepository
from packages.v1.administrativo.schemas.c_caixa_item_schema import \
CaixaItemSchema
class UpdateAction(BaseAction): class UpdateAction(BaseAction):
def execute(self, caixa_item_id : int, caixa_item_schema : CaixaItemSchema): def execute(self, caixa_item_id: int, caixa_item_schema: CaixaItemSchema):
# Instância o repositório desejado # Instância o repositório desejado
update = UpdateRepository() update = UpdateRepository()

View file

@ -1,10 +1,12 @@
from packages.v1.administrativo.schemas.c_caixa_servico_schema import CCaixaServicoIdSchema from packages.v1.administrativo.repositories.c_caixa_servico.c_caixa_servico_delete_repository import \
from packages.v1.administrativo.repositories.c_caixa_servico.c_caixa_servico_delete_repository import DeleteRepository DeleteRepository
from packages.v1.administrativo.schemas.c_caixa_servico_schema import \
CCaixaServicoIdSchema
class DeleteAction: class DeleteAction:
def execute(self, usuario_schema : CCaixaServicoIdSchema): def execute(self, usuario_schema: CCaixaServicoIdSchema):
delete_repository = DeleteRepository() delete_repository = DeleteRepository()

View file

@ -1,10 +1,13 @@
from abstracts.action import BaseAction from abstracts.action import BaseAction
from packages.v1.administrativo.schemas.c_caixa_servico_schema import CCaixaServicoDescricaoSchema from packages.v1.administrativo.repositories.c_caixa_servico.c_caixa_servico_get_by_descricao_repository import \
from packages.v1.administrativo.repositories.c_caixa_servico.c_caixa_servico_get_by_descricao_repository import ShowRepository ShowRepository
from packages.v1.administrativo.schemas.c_caixa_servico_schema import \
CCaixaServicoDescricaoSchema
class ShowAction(BaseAction): class ShowAction(BaseAction):
def execute(self, caixa_servico_schema : CCaixaServicoDescricaoSchema): def execute(self, caixa_servico_schema: CCaixaServicoDescricaoSchema):
# Instânciamento do repositório sql # Instânciamento do repositório sql
show_repository = ShowRepository() show_repository = ShowRepository()

View file

@ -1,16 +0,0 @@
from abstracts.action import BaseAction
from packages.v1.administrativo.schemas.c_caixa_servico_schema import CCaixaServicoSistemaIdSchema
from packages.v1.administrativo.repositories.c_caixa_servico.c_caixa_servico_get_by_sistema_id_repository import ShowSistemaIdRepository
class ShowSistemaIdAction(BaseAction):
def execute(self, caixa_servico_schema : CCaixaServicoSistemaIdSchema):
# Instânciamento do repositório sql
show_sistema_id_repository = ShowSistemaIdRepository()
# Execução do sql
response = show_sistema_id_repository.execute(caixa_servico_schema)
# Retorno da informação
return response

View file

@ -1,7 +1,6 @@
from abstracts.action import BaseAction from abstracts.action import BaseAction
from packages.v1.administrativo.repositories.c_caixa_servico.c_caixa_servico_index_repository import ( from packages.v1.administrativo.repositories.c_caixa_servico.c_caixa_servico_index_repository import \
IndexRepository, IndexRepository
)
class IndexAction(BaseAction): class IndexAction(BaseAction):

View file

@ -1,9 +1,7 @@
from packages.v1.administrativo.schemas.c_caixa_servico_schema import ( from packages.v1.administrativo.repositories.c_caixa_servico.c_caixa_servico_save_repository import \
CCaixaServicoSaveSchema, SaveRepository
) from packages.v1.administrativo.schemas.c_caixa_servico_schema import \
from packages.v1.administrativo.repositories.c_caixa_servico.c_caixa_servico_save_repository import ( CCaixaServicoSaveSchema
SaveRepository,
)
class SaveAction: class SaveAction:

View file

@ -1,10 +1,13 @@
from abstracts.action import BaseAction from abstracts.action import BaseAction
from packages.v1.administrativo.schemas.c_caixa_servico_schema import CCaixaServicoSchema from packages.v1.administrativo.repositories.c_caixa_servico.c_caixa_servico_show_repository import \
from packages.v1.administrativo.repositories.c_caixa_servico.c_caixa_servico_show_repository import ShowRepository ShowRepository
from packages.v1.administrativo.schemas.c_caixa_servico_schema import \
CCaixaServicoSchema
class ShowAction(BaseAction): class ShowAction(BaseAction):
def execute(self, usuario_schema : CCaixaServicoSchema): def execute(self, usuario_schema: CCaixaServicoSchema):
# Instânciamento do repositório sql # Instânciamento do repositório sql
show_repository = ShowRepository() show_repository = ShowRepository()

View file

@ -1,10 +1,14 @@
from packages.v1.administrativo.schemas.c_caixa_servico_schema import CCaixaServicoUpdateSchema from packages.v1.administrativo.repositories.c_caixa_servico.c_caixa_servico_update_repository import \
from packages.v1.administrativo.repositories.c_caixa_servico.c_caixa_servico_update_repository import UpdateRepository UpdateRepository
from packages.v1.administrativo.schemas.c_caixa_servico_schema import \
CCaixaServicoUpdateSchema
class UpdateAction: class UpdateAction:
def execute(self, caixa_servico_id : int, c_caixa_servico_schema : CCaixaServicoUpdateSchema): def execute(
self, caixa_servico_id: int, c_caixa_servico_schema: CCaixaServicoUpdateSchema
):
save_repository = UpdateRepository() save_repository = UpdateRepository()

View file

@ -1,35 +0,0 @@
from abstracts.action import BaseAction
from packages.v1.administrativo.repositories.g_cartorio.g_cartorio_delete_repository import (
GCartorioDeleteRepository,
)
from packages.v1.administrativo.schemas.g_cartorio_schema import GCartorioIdSchema
class GCartorioDeleteAction(BaseAction):
"""
Serviço responsável por encapsular a lógica de negócio para a operação
de exclusão de um registro na tabela G_CARTORIO.
"""
def execute(self, g_cartorio_id_schema: GCartorioIdSchema):
"""
Executa a operação de exclusão no banco de dados.
Args:
g_cartorio_id_schema (GCartorioIdSchema):
O esquema contendo o ID do registro a ser excluído.
Returns:
O resultado da operação de exclusão.
"""
# ----------------------------------------------------
# Instanciamento do repositório
# ----------------------------------------------------
g_cartorio_delete_repository = GCartorioDeleteRepository()
# ----------------------------------------------------
# Execução da exclusão
# ----------------------------------------------------
response = g_cartorio_delete_repository.execute(g_cartorio_id_schema)
return response

View file

@ -1,37 +0,0 @@
from abstracts.action import BaseAction
from packages.v1.administrativo.repositories.g_cartorio.g_cartorio_index_repository import (
GCartorioIndexRepository,
)
class GCartorioIndexAction(BaseAction):
"""
Serviço responsável por encapsular a lógica de negócio para a operação
de listagem de todos os registros na tabela G_NATUREZA_TITULO.
"""
def execute(self):
"""
Executa a operação de listagem no banco de dados.
Args:
g_cartorio_index_schema (GCartorioIndexSchema):
Esquema contendo parâmetros opcionais de filtro.
Returns:
A lista de registros encontrados.
"""
# ----------------------------------------------------
# Instanciamento do repositório
# ----------------------------------------------------
g_cartorio_index_repository = GCartorioIndexRepository()
# ----------------------------------------------------
# Execução do repositório
# ----------------------------------------------------
response = g_cartorio_index_repository.execute()
# ----------------------------------------------------
# Retorno da informação
# --------------------------------
return response

View file

@ -1,38 +0,0 @@
from abstracts.action import BaseAction
from packages.v1.administrativo.repositories.g_cartorio.g_cartorio_save_repository import (
GCartorioSaveRepository,
)
from packages.v1.administrativo.schemas.g_cartorio_schema import GCartorioSaveSchema
class GCartorioSaveAction(BaseAction):
"""
Serviço responsável por encapsular a lógica de negócio para a operação
de salvamento de um novo registro na tabela G_GRAMATICA.
"""
def execute(self, g_cartorio_save_schema: GCartorioSaveSchema):
"""
Executa a operação de salvamento.
Args:
g_cartorio_schema (GCartorioSchema):
O esquema com os dados a serem persistidos.
Returns:
O resultado da operação de salvamento.
"""
# ----------------------------------------------------
# Instanciamento do repositório
# ----------------------------------------------------
g_cartorio_save_repository = GCartorioSaveRepository()
# ----------------------------------------------------
# Execução do repositório
# ----------------------------------------------------
response = g_cartorio_save_repository.execute(g_cartorio_save_schema)
# ----------------------------------------------------
# Retorno da informação
# ----------------------------------------------------
return response

View file

@ -1,38 +0,0 @@
from abstracts.action import BaseAction
from packages.v1.administrativo.repositories.g_cartorio.g_cartorio_show_repository import (
GCartorioShowRepository,
)
from packages.v1.administrativo.schemas.g_cartorio_schema import GCartorioIdSchema
class GCartorioShowAction(BaseAction):
"""
Serviço responsável por encapsular a lógica de negócio para a exibição
de um registro na tabela G_NATUREZA_TITULO.
"""
def execute(self, g_cartorio_id_schema: GCartorioIdSchema):
"""
Executa a operação de exibição.
Args:
g_cartorio_id_schema (GCartorioIdSchema):
O esquema com o ID do registro a ser exibido.
Returns:
O resultado da operação de exibição.
"""
# ----------------------------------------------------
# Instanciamento do repositório
# ----------------------------------------------------
g_cartorio_show_repository = GCartorioShowRepository()
# ----------------------------------------------------
# Execução do repositório
# ----------------------------------------------------
response = g_cartorio_show_repository.execute(g_cartorio_id_schema)
# ----------------------------------------------------
# Retorno da informação
# ----------------------------------------------------
return response

View file

@ -1,38 +0,0 @@
from abstracts.action import BaseAction
from packages.v1.administrativo.repositories.g_cartorio.g_cartorio_update_repository import (
GCartorioUpdateRepository,
)
from packages.v1.administrativo.schemas.g_cartorio_schema import GCartorioUpdateSchema
class GCartorioUpdateAction(BaseAction):
"""
Serviço responsável por encapsular a lógica de negócio para a atualização
de um registro na tabela G_NATUREZA_TITULO.
"""
def execute(self, g_cartorio_update_schema: GCartorioUpdateSchema):
"""
Executa a operação de atualização.
Args:
g_cartorio_update_schema (GCartorioUpdateSchema):
O esquema com os dados a serem atualizados.
Returns:
O resultado da operação de atualização.
"""
# ----------------------------------------------------
# Instanciamento do repositório de atualização
# ----------------------------------------------------
g_cartorio_update_repository = GCartorioUpdateRepository()
# ----------------------------------------------------
# Execução do repositório
# ----------------------------------------------------
response = g_cartorio_update_repository.execute(g_cartorio_update_schema)
# ----------------------------------------------------
# Retorno do resultado
# ----------------------------------------------------
return response

View file

@ -1,6 +1,7 @@
from abstracts.action import BaseAction from abstracts.action import BaseAction
from packages.v1.administrativo.repositories.g_cidade.g_cidade_delete_repository import \
DeleteRepository
from packages.v1.administrativo.schemas.g_cidade_schema import GCidadeIdSchema from packages.v1.administrativo.schemas.g_cidade_schema import GCidadeIdSchema
from packages.v1.administrativo.repositories.g_cidade.g_cidade_delete_repository import DeleteRepository
class DeleteAction(BaseAction): class DeleteAction(BaseAction):

View file

@ -1,6 +1,8 @@
from abstracts.action import BaseAction from abstracts.action import BaseAction
from packages.v1.administrativo.schemas.g_cidade_schema import GCidadeNomeSchema from packages.v1.administrativo.repositories.g_cidade.g_cidade_get_by_nome_repository import \
from packages.v1.administrativo.repositories.g_cidade.g_cidade_get_by_nome_repository import GetByNomeRepository GetByNomeRepository
from packages.v1.administrativo.schemas.g_cidade_schema import \
GCidadeNomeSchema
class GetByNomeAction(BaseAction): class GetByNomeAction(BaseAction):

View file

@ -1,5 +1,7 @@
from abstracts.action import BaseAction from abstracts.action import BaseAction
from packages.v1.administrativo.repositories.g_cidade.g_cidade_index_repository import IndexRepository from packages.v1.administrativo.repositories.g_cidade.g_cidade_index_repository import \
IndexRepository
class IndexAction(BaseAction): class IndexAction(BaseAction):
""" """

View file

@ -1,6 +1,9 @@
from abstracts.action import BaseAction from abstracts.action import BaseAction
from packages.v1.administrativo.schemas.g_cidade_schema import GCidadeSaveSchema from packages.v1.administrativo.repositories.g_cidade.g_cidade_save_repository import \
from packages.v1.administrativo.repositories.g_cidade.g_cidade_save_repository import SaveRepository SaveRepository
from packages.v1.administrativo.schemas.g_cidade_schema import \
GCidadeSaveSchema
class SaveAction(BaseAction): class SaveAction(BaseAction):
""" """

View file

@ -1,6 +1,8 @@
from abstracts.action import BaseAction from abstracts.action import BaseAction
from packages.v1.administrativo.repositories.g_cidade.g_cidade_show_repository import \
ShowRepository
from packages.v1.administrativo.schemas.g_cidade_schema import GCidadeIdSchema from packages.v1.administrativo.schemas.g_cidade_schema import GCidadeIdSchema
from packages.v1.administrativo.repositories.g_cidade.g_cidade_show_repository import ShowRepository
class ShowAction(BaseAction): class ShowAction(BaseAction):
""" """

View file

@ -1,11 +1,13 @@
from packages.v1.administrativo.schemas.g_cidade_schema import GCidadeUpdateSchema from packages.v1.administrativo.repositories.g_cidade.g_cidade_update_repository import \
from packages.v1.administrativo.repositories.g_cidade.g_cidade_update_repository import UpdateRepository UpdateRepository
from packages.v1.administrativo.schemas.g_cidade_schema import \
GCidadeUpdateSchema
class UpdateAction: class UpdateAction:
""" """
Serviço responsável por encapsular a lógica de negócio para a atualização Serviço responsável por encapsular a lógica de negócio para a atualização
de um registro na tabela G_CIDADE. de um registro na tabela G_CIDADE.
""" """
def execute(self, cidade_id: int, g_cidade_schema: GCidadeUpdateSchema): def execute(self, cidade_id: int, g_cidade_schema: GCidadeUpdateSchema):

View file

@ -1,35 +0,0 @@
from abstracts.action import BaseAction
from packages.v1.administrativo.repositories.g_emolumento.g_emolumento_delete_repository import (
GEmolumentoDeleteRepository,
)
from packages.v1.administrativo.schemas.g_emolumento_schema import GEmolumentoIdSchema
class GEmolumentoDeleteAction(BaseAction):
"""
Serviço responsável por encapsular a lógica de negócio para a operação
de exclusão de um registro na tabela G_EMOLUMENTO.
"""
def execute(self, g_emolumento_id_schema: GEmolumentoIdSchema):
"""
Executa a operação de exclusão no banco de dados.
Args:
g_emolumento_id_schema (GEmolumentoIdSchema):
O esquema contendo o ID do registro a ser excluído.
Returns:
O resultado da operação de exclusão.
"""
# ----------------------------------------------------
# Instanciamento do repositório
# ----------------------------------------------------
g_emolumento_delete_repository = GEmolumentoDeleteRepository()
# ----------------------------------------------------
# Execução da exclusão
# ----------------------------------------------------
response = g_emolumento_delete_repository.execute(g_emolumento_id_schema)
return response

View file

@ -1,44 +0,0 @@
from abstracts.action import BaseAction
from packages.v1.administrativo.repositories.g_emolumento.g_emolumento_index_by_sistema_id_repository import (
GEmolumentoIndexBySistemaIdRepository,
)
from packages.v1.administrativo.schemas.g_emolumento_schema import (
GEmolumentoSistemaIdSchema,
)
class GEmolumentoIndexBySistemaIdAction(BaseAction):
"""
Serviço responsável por encapsular a lógica de negócio para a exibição
de um registro na tabela G_NATUREZA_TITULO.
"""
def execute(self, g_emolumento_sistema_id_schema: GEmolumentoSistemaIdSchema):
"""
Executa a operação de exibição.
Args:
g_emolumento_id_schema (GEmolumentoIdSchema):
O esquema com o ID do registro a ser exibido.
Returns:
O resultado da operação de exibição.
"""
# ----------------------------------------------------
# Instanciamento do repositório
# ----------------------------------------------------
g_emolumento_index_by_sistema_id_repository = (
GEmolumentoIndexBySistemaIdRepository()
)
# ----------------------------------------------------
# Execução do repositório
# ----------------------------------------------------
response = g_emolumento_index_by_sistema_id_repository.execute(
g_emolumento_sistema_id_schema
)
# ----------------------------------------------------
# Retorno da informação
# ----------------------------------------------------
return response

View file

@ -1,38 +0,0 @@
from abstracts.action import BaseAction
from packages.v1.administrativo.repositories.g_emolumento.g_emolumento_save_repository import (
GEmolumentoSaveRepository,
)
from packages.v1.administrativo.schemas.g_emolumento_schema import GEmolumentoSaveSchema
class GEmolumentoSaveAction(BaseAction):
"""
Serviço responsável por encapsular a lógica de negócio para a operação
de salvamento de um novo registro na tabela G_EMOLUMENTO.
"""
def execute(self, g_emolumento_save_schema: GEmolumentoSaveSchema):
"""
Executa a operação de salvamento.
Args:
g_emolumento_schema (GEmolumentoSchema):
O esquema com os dados a serem persistidos.
Returns:
O resultado da operação de salvamento.
"""
# ----------------------------------------------------
# Instanciamento do repositório
# ----------------------------------------------------
g_emolumento_save_repository = GEmolumentoSaveRepository()
# ----------------------------------------------------
# Execução do repositório
# ----------------------------------------------------
response = g_emolumento_save_repository.execute(g_emolumento_save_schema)
# ----------------------------------------------------
# Retorno da informação
# ----------------------------------------------------
return response

View file

@ -1,38 +0,0 @@
from abstracts.action import BaseAction
from packages.v1.administrativo.repositories.g_emolumento.g_emolumento_show_repository import (
GEmolumentoShowRepository,
)
from packages.v1.administrativo.schemas.g_emolumento_schema import GEmolumentoIdSchema
class GEmolumentoShowAction(BaseAction):
"""
Serviço responsável por encapsular a lógica de negócio para a exibição
de um registro na tabela G_NATUREZA_TITULO.
"""
def execute(self, g_emolumento_id_schema: GEmolumentoIdSchema):
"""
Executa a operação de exibição.
Args:
g_emolumento_id_schema (GEmolumentoIdSchema):
O esquema com o ID do registro a ser exibido.
Returns:
O resultado da operação de exibição.
"""
# ----------------------------------------------------
# Instanciamento do repositório
# ----------------------------------------------------
g_emolumento_show_repository = GEmolumentoShowRepository()
# ----------------------------------------------------
# Execução do repositório
# ----------------------------------------------------
response = g_emolumento_show_repository.execute(g_emolumento_id_schema)
# ----------------------------------------------------
# Retorno da informação
# ----------------------------------------------------
return response

View file

@ -1,40 +0,0 @@
from abstracts.action import BaseAction
from packages.v1.administrativo.repositories.g_emolumento.g_emolumento_update_repository import (
GEmolumentoUpdateRepository,
)
from packages.v1.administrativo.schemas.g_emolumento_schema import (
GEmolumentoUpdateSchema,
)
class GEmolumentoUpdateAction(BaseAction):
"""
Serviço responsável por encapsular a lógica de negócio para a atualização
de um registro na tabela G_NATUREZA_TITULO.
"""
def execute(self, g_emolumento_update_schema: GEmolumentoUpdateSchema):
"""
Executa a operação de atualização.
Args:
g_emolumento_update_schema (GEmolumentoUpdateSchema):
O esquema com os dados a serem atualizados.
Returns:
O resultado da operação de atualização.
"""
# ----------------------------------------------------
# Instanciamento do repositório de atualização
# ----------------------------------------------------
g_emolumento_update_repository = GEmolumentoUpdateRepository()
# ----------------------------------------------------
# Execução do repositório
# ----------------------------------------------------
response = g_emolumento_update_repository.execute(g_emolumento_update_schema)
# ----------------------------------------------------
# Retorno do resultado
# ----------------------------------------------------
return response

View file

@ -1,39 +0,0 @@
from abstracts.action import BaseAction
from packages.v1.administrativo.repositories.g_emolumento_item.g_emolumento_item_delete_repository import (
GEmolumentoItemDeleteRepository,
)
from packages.v1.administrativo.schemas.g_emolumento_item_schema import (
GEmolumentoItemIdSchema,
)
class GEmolumentoItemDeleteAction(BaseAction):
"""
Serviço responsável por encapsular a lógica de negócio para a operação
de exclusão de um registro na tabela G_EMOLUMENTO_ITEM.
"""
def execute(self, g_emolumento_item_id_schema: GEmolumentoItemIdSchema):
"""
Executa a operação de exclusão no banco de dados.
Args:
g_emolumento_item_id_schema (GEmolumentoItemIdSchema):
O esquema contendo o ID do registro a ser excluído.
Returns:
O resultado da operação de exclusão.
"""
# ----------------------------------------------------
# Instanciamento do repositório
# ----------------------------------------------------
g_emolumento_item_delete_repository = GEmolumentoItemDeleteRepository()
# ----------------------------------------------------
# Execução da exclusão
# ----------------------------------------------------
response = g_emolumento_item_delete_repository.execute(
g_emolumento_item_id_schema
)
return response

View file

@ -1,74 +0,0 @@
from __future__ import annotations
from decimal import Decimal
from typing import Iterable, Optional, Protocol, Any
class EmolumentoItemLike(Protocol):
"""Protocolo mínimo esperado para um item de emolumento."""
valor_inicio: Optional[Decimal]
valor_fim: Optional[Decimal]
class GEmolumentoItemGetFaixaValorAction:
"""
Responsável por escolher, dentre uma coleção de itens de emolumento,
aquele cuja faixa [valor_inicio, valor_fim] contem o valor_documento.
- Se houver mais de um candidato, prioriza o de maior valor_inicio (faixa mais específica).
- Se não houver faixa que contenha o valor_documento, tenta a faixa 'aberta' (valor_fim nulo).
- Persistindo a ausência, retorna o item cujo valor_inicio é o mais próximo abaixo do valor_documento.
"""
@staticmethod
def _para_decimal(valor: Any, padrao: str = "0") -> Decimal:
return valor if isinstance(valor, Decimal) else Decimal(str(valor or padrao))
def execute(
self,
itens: Iterable[EmolumentoItemLike],
valor_documento: Decimal,
) -> EmolumentoItemLike:
lista = list(itens)
if not lista:
raise ValueError("Nenhum item de emolumento foi informado.")
valor_doc = self._para_decimal(valor_documento)
candidatos: list[tuple[Decimal, Decimal, EmolumentoItemLike]] = []
abertos: list[tuple[Decimal, EmolumentoItemLike]] = []
abaixo: list[tuple[Decimal, EmolumentoItemLike]] = []
for item in lista:
ini = self._para_decimal(getattr(item, "valor_inicio", None))
fim_raw = getattr(item, "valor_fim", None)
fim = (
self._para_decimal(fim_raw, padrao="Infinity")
if fim_raw is not None
else Decimal("Infinity")
)
if ini <= valor_doc <= fim:
candidatos.append((ini, fim, item))
elif fim == Decimal("Infinity") and ini <= valor_doc:
abertos.append((ini, item))
elif ini <= valor_doc:
abaixo.append((ini, item))
if candidatos:
candidatos.sort(key=lambda t: (t[0], t[1])) # maior ini e menor fim
return candidatos[-1][2]
if abertos:
abertos.sort(key=lambda t: t[0]) # maior ini
return abertos[-1][1]
if abaixo:
abaixo.sort(key=lambda t: t[0]) # maior ini
return abaixo[-1][1]
# Fallback: não há faixa adequada nem valores abaixo; devolve o de menor valor_inicio
lista_ordenada = sorted(
lista, key=lambda it: self._para_decimal(getattr(it, "valor_inicio", None))
)
return lista_ordenada[0]

View file

@ -1,44 +0,0 @@
from abstracts.action import BaseAction
from packages.v1.administrativo.repositories.g_emolumento_item.g_emolumento_item_index_repository import (
GEmolumentoItemIndexRepository,
)
from packages.v1.administrativo.schemas.g_emolumento_item_schema import (
GEmolumentoItemIndexSchema,
)
class GEmolumentoItemIndexAction(BaseAction):
"""
Serviço responsável por encapsular a lógica de negócio para a operação
de listagem de todos os registros na tabela G_NATUREZA_TITULO.
"""
def execute(
self, g_emolumento_item_emolumento_index_schema: GEmolumentoItemIndexSchema
):
"""
Executa a operação de listagem no banco de dados.
Args:
g_emolumento_item_index_schema (GEmolumentoItemIndexSchema):
Esquema contendo parâmetros opcionais de filtro.
Returns:
A lista de registros encontrados.
"""
# ----------------------------------------------------
# Instanciamento do repositório
# ----------------------------------------------------
g_emolumento_item_index_repository = GEmolumentoItemIndexRepository()
# ----------------------------------------------------
# Execução do repositório
# ----------------------------------------------------
response = g_emolumento_item_index_repository.execute(
g_emolumento_item_emolumento_index_schema
)
# ----------------------------------------------------
# Retorno da informação
# --------------------------------
return response

View file

@ -1,36 +0,0 @@
from abstracts.action import BaseAction
from packages.v1.administrativo.repositories.g_emolumento_item.g_emolumento_item_save_repository import GEmolumentoItemSaveRepository
from packages.v1.administrativo.schemas.g_emolumento_item_schema import GEmolumentoItemSaveSchema
class GEmolumentoItemSaveAction(BaseAction):
"""
Serviço responsável por encapsular a lógica de negócio para a operação
de salvamento de um novo registro na tabela G_EMOLUMENTO_ITEM.
"""
def execute(self, g_emolumento_item_save_schema: GEmolumentoItemSaveSchema):
"""
Executa a operação de salvamento.
Args:
g_emolumento_item_schema (GEmolumentoItemSchema):
O esquema com os dados a serem persistidos.
Returns:
O resultado da operação de salvamento.
"""
# ----------------------------------------------------
# Instanciamento do repositório
# ----------------------------------------------------
g_emolumento_item_save_repository = GEmolumentoItemSaveRepository()
# ----------------------------------------------------
# Execução do repositório
# ----------------------------------------------------
response = g_emolumento_item_save_repository.execute(g_emolumento_item_save_schema)
# ----------------------------------------------------
# Retorno da informação
# ----------------------------------------------------
return response

View file

@ -1,42 +0,0 @@
from abstracts.action import BaseAction
from packages.v1.administrativo.repositories.g_emolumento_item.g_emolumento_item_show_repository import (
GEmolumentoItemShowRepository,
)
from packages.v1.administrativo.schemas.g_emolumento_item_schema import (
GEmolumentoItemIdSchema,
)
class GEmolumentoItemShowAction(BaseAction):
"""
Serviço responsável por encapsular a lógica de negócio para a exibição
de um registro na tabela G_NATUREZA_TITULO.
"""
def execute(self, g_emolumento_item_id_schema: GEmolumentoItemIdSchema):
"""
Executa a operação de exibição.
Args:
g_emolumento_item_id_schema (GEmolumentoItemIdSchema):
O esquema com o ID do registro a ser exibido.
Returns:
O resultado da operação de exibição.
"""
# ----------------------------------------------------
# Instanciamento do repositório
# ----------------------------------------------------
g_emolumento_item_show_repository = GEmolumentoItemShowRepository()
# ----------------------------------------------------
# Execução do repositório
# ----------------------------------------------------
response = g_emolumento_item_show_repository.execute(
g_emolumento_item_id_schema
)
# ----------------------------------------------------
# Retorno da informação
# ----------------------------------------------------
return response

View file

@ -1,42 +0,0 @@
from abstracts.action import BaseAction
from packages.v1.administrativo.repositories.g_emolumento_item.g_emolumento_item_update_repository import (
GEmolumentoItemUpdateRepository,
)
from packages.v1.administrativo.schemas.g_emolumento_item_schema import (
GEmolumentoItemUpdateSchema,
)
class GEmolumentoItemUpdateAction(BaseAction):
"""
Serviço responsável por encapsular a lógica de negócio para a atualização
de um registro na tabela G_NATUREZA_TITULO.
"""
def execute(self, g_emolumento_item_update_schema: GEmolumentoItemUpdateSchema):
"""
Executa a operação de atualização.
Args:
g_emolumento_item_update_schema (GEmolumentoItemUpdateSchema):
O esquema com os dados a serem atualizados.
Returns:
O resultado da operação de atualização.
"""
# ----------------------------------------------------
# Instanciamento do repositório de atualização
# ----------------------------------------------------
g_emolumento_item_update_repository = GEmolumentoItemUpdateRepository()
# ----------------------------------------------------
# Execução do repositório
# ----------------------------------------------------
response = g_emolumento_item_update_repository.execute(
g_emolumento_item_update_schema
)
# ----------------------------------------------------
# Retorno do resultado
# ----------------------------------------------------
return response

View file

@ -1,44 +0,0 @@
from abstracts.action import BaseAction
# O Schema precisa ser adaptado para GEmolumentoItem, talvez recebendo um ID de Emolumento
# ou um ID do próprio Item, dependendo da necessidade de listagem.
# Vamos sugerir um Schema específico para a listagem (Index) que pode receber um Emolumento ID
# para listar todos os seus itens, mantendo o padrão do arquivo original que usava um 'SistemaIdSchema'.
# Para simplificar, vamos assumir um schema de filtro ou um schema base para Index.
# Sugerimos a criação de:
# from packages.v1.administrativo.schemas.g_emolumento_item_schema import GEmolumentoItemIndexSchema
# (ou GEmolumentoItemEmolumentoIdSchema se for o padrão da aplicação)
from packages.v1.administrativo.schemas.g_emolumento_item_schema import GEmolumentoItemValorSchema
# O repositório ValorRepository deve ser substituído pelo GEmolumentoItemValorRepository.
from packages.v1.administrativo.repositories.g_emolumento_item.g_emolumento_item_valor_repository import ValorRepository
class ValorAction(BaseAction):
"""
Serviço responsável por encapsular a lógica de negócio para a operação
de listagem de todos os registros na tabela G_EMOLUMENTO_ITEM,
utilizando a DDL fornecida.
"""
# Mantendo o padrão de nome de método do arquivo original
def execute(self, emolumento_item_schema: GEmolumentoItemValorSchema):
"""
Executa a operação de listagem de G_EMOLUMENTO_ITEM no banco de dados.
Args:
emolumento_item_schema: Esquema com parâmetros de filtro/listagem
(por exemplo, ID do Emolumento pai, se a listagem for
filtrada por ele, ou parâmetros de paginação).
Returns:
A lista de todos os registros de G_EMOLUMENTO_ITEM que satisfazem o filtro.
"""
# Instanciamento do repositório
# O nome do repositório foi adaptado com o prefixo 'GEmolumentoItem'
index_repository = ValorRepository()
# Execução do repositório
# O nome do parâmetro foi adaptado para 'emolumento_item_schema'
response = index_repository.execute(emolumento_item_schema)
# Retorno da informação
return response

View file

@ -1,39 +0,0 @@
from abstracts.action import BaseAction
from packages.v1.administrativo.repositories.g_emolumento_periodo.g_emolumento_periodo_delete_repository import (
GEmolumentoPeriodoDeleteRepository,
)
from packages.v1.administrativo.schemas.g_emolumento_periodo_schema import (
GEmolumentoPeriodoIdSchema,
)
class GEmolumentoPeriodoDeleteAction(BaseAction):
"""
Serviço responsável por encapsular a lógica de negócio para a operação
de exclusão de um registro na tabela G_EMOLUMENTO_PERIODO.
"""
def execute(self, g_emolumento_periodo_id_schema: GEmolumentoPeriodoIdSchema):
"""
Executa a operação de exclusão no banco de dados.
Args:
g_emolumento_periodo_id_schema (GEmolumentoPeriodoIdSchema):
O esquema contendo o ID do registro a ser excluído.
Returns:
O resultado da operação de exclusão.
"""
# ----------------------------------------------------
# Instanciamento do repositório
# ----------------------------------------------------
g_emolumento_periodo_delete_repository = GEmolumentoPeriodoDeleteRepository()
# ----------------------------------------------------
# Execução da exclusão
# ----------------------------------------------------
response = g_emolumento_periodo_delete_repository.execute(
g_emolumento_periodo_id_schema
)
return response

View file

@ -1,37 +0,0 @@
from abstracts.action import BaseAction
from packages.v1.administrativo.repositories.g_emolumento_periodo.g_emolumento_periodo_index_repository import (
GEmolumentoPeriodoIndexRepository,
)
class GEmolumentoPeriodoIndexAction(BaseAction):
"""
Serviço responsável por encapsular a lógica de negócio para a operação
de listagem de todos os registros na tabela G_NATUREZA_TITULO.
"""
def execute(self):
"""
Executa a operação de listagem no banco de dados.
Args:
g_emolumento_periodo_index_schema (GEmolumentoPeriodoIndexSchema):
Esquema contendo parâmetros opcionais de filtro.
Returns:
A lista de registros encontrados.
"""
# ----------------------------------------------------
# Instanciamento do repositório
# ----------------------------------------------------
g_emolumento_periodo_index_repository = GEmolumentoPeriodoIndexRepository()
# ----------------------------------------------------
# Execução do repositório
# ----------------------------------------------------
response = g_emolumento_periodo_index_repository.execute()
# ----------------------------------------------------
# Retorno da informação
# --------------------------------
return response

View file

@ -1,42 +0,0 @@
from abstracts.action import BaseAction
from packages.v1.administrativo.repositories.g_emolumento_periodo.g_emolumento_periodo_save_repository import (
GEmolumentoPeriodoSaveRepository,
)
from packages.v1.administrativo.schemas.g_emolumento_periodo_schema import (
GEmolumentoPeriodoSaveSchema,
)
class GEmolumentoPeriodoSaveAction(BaseAction):
"""
Serviço responsável por encapsular a lógica de negócio para a operação
de salvamento de um novo registro na tabela G_GRAMATICA.
"""
def execute(self, g_emolumento_periodo_save_schema: GEmolumentoPeriodoSaveSchema):
"""
Executa a operação de salvamento.
Args:
g_emolumento_periodo_schema (GEmolumentoPeriodoSchema):
O esquema com os dados a serem persistidos.
Returns:
O resultado da operação de salvamento.
"""
# ----------------------------------------------------
# Instanciamento do repositório
# ----------------------------------------------------
g_emolumento_periodo_save_repository = GEmolumentoPeriodoSaveRepository()
# ----------------------------------------------------
# Execução do repositório
# ----------------------------------------------------
response = g_emolumento_periodo_save_repository.execute(
g_emolumento_periodo_save_schema
)
# ----------------------------------------------------
# Retorno da informação
# ----------------------------------------------------
return response

View file

@ -1,42 +0,0 @@
from abstracts.action import BaseAction
from packages.v1.administrativo.repositories.g_emolumento_periodo.g_emolumento_periodo_show_repository import (
GEmolumentoPeriodoShowRepository,
)
from packages.v1.administrativo.schemas.g_emolumento_periodo_schema import (
GEmolumentoPeriodoIdSchema,
)
class GEmolumentoPeriodoShowAction(BaseAction):
"""
Serviço responsável por encapsular a lógica de negócio para a exibição
de um registro na tabela G_NATUREZA_TITULO.
"""
def execute(self, g_emolumento_periodo_id_schema: GEmolumentoPeriodoIdSchema):
"""
Executa a operação de exibição.
Args:
g_emolumento_periodo_id_schema (GEmolumentoPeriodoIdSchema):
O esquema com o ID do registro a ser exibido.
Returns:
O resultado da operação de exibição.
"""
# ----------------------------------------------------
# Instanciamento do repositório
# ----------------------------------------------------
g_emolumento_periodo_show_repository = GEmolumentoPeriodoShowRepository()
# ----------------------------------------------------
# Execução do repositório
# ----------------------------------------------------
response = g_emolumento_periodo_show_repository.execute(
g_emolumento_periodo_id_schema
)
# ----------------------------------------------------
# Retorno da informação
# ----------------------------------------------------
return response

View file

@ -1,44 +0,0 @@
from abstracts.action import BaseAction
from packages.v1.administrativo.repositories.g_emolumento_periodo.g_emolumento_periodo_update_repository import (
GEmolumentoPeriodoUpdateRepository,
)
from packages.v1.administrativo.schemas.g_emolumento_periodo_schema import (
GEmolumentoPeriodoUpdateSchema,
)
class GEmolumentoPeriodoUpdateAction(BaseAction):
"""
Serviço responsável por encapsular a lógica de negócio para a atualização
de um registro na tabela G_NATUREZA_TITULO.
"""
def execute(
self, g_emolumento_periodo_update_schema: GEmolumentoPeriodoUpdateSchema
):
"""
Executa a operação de atualização.
Args:
g_emolumento_periodo_update_schema (GEmolumentoPeriodoUpdateSchema):
O esquema com os dados a serem atualizados.
Returns:
O resultado da operação de atualização.
"""
# ----------------------------------------------------
# Instanciamento do repositório de atualização
# ----------------------------------------------------
g_emolumento_periodo_update_repository = GEmolumentoPeriodoUpdateRepository()
# ----------------------------------------------------
# Execução do repositório
# ----------------------------------------------------
response = g_emolumento_periodo_update_repository.execute(
g_emolumento_periodo_update_schema
)
# ----------------------------------------------------
# Retorno do resultado
# ----------------------------------------------------
return response

View file

@ -1,35 +0,0 @@
from abstracts.action import BaseAction
from packages.v1.administrativo.repositories.g_gramatica.g_gramatica_delete_repository import (
GGramaticaDeleteRepository,
)
from packages.v1.administrativo.schemas.g_gramatica_schema import GGramaticaIdSchema
class GGramaticaDeleteAction(BaseAction):
"""
Serviço responsável por encapsular a lógica de negócio para a operação
de exclusão de um registro na tabela G_GRAMATICA.
"""
def execute(self, g_gramatica_id_schema: GGramaticaIdSchema):
"""
Executa a operação de exclusão no banco de dados.
Args:
g_gramatica_id_schema (GGramaticaIdSchema):
O esquema contendo o ID do registro a ser excluído.
Returns:
O resultado da operação de exclusão.
"""
# ----------------------------------------------------
# Instanciamento do repositório
# ----------------------------------------------------
g_gramatica_delete_repository = GGramaticaDeleteRepository()
# ----------------------------------------------------
# Execução da exclusão
# ----------------------------------------------------
response = g_gramatica_delete_repository.execute(g_gramatica_id_schema)
return response

View file

@ -1,37 +0,0 @@
from abstracts.action import BaseAction
from packages.v1.administrativo.repositories.g_gramatica.g_gramatica_index_repository import (
GGramaticaIndexRepository,
)
class GGramaticaIndexAction(BaseAction):
"""
Serviço responsável por encapsular a lógica de negócio para a operação
de listagem de todos os registros na tabela G_NATUREZA_TITULO.
"""
def execute(self):
"""
Executa a operação de listagem no banco de dados.
Args:
g_gramatica_index_schema (GGramaticaIndexSchema):
Esquema contendo parâmetros opcionais de filtro.
Returns:
A lista de registros encontrados.
"""
# ----------------------------------------------------
# Instanciamento do repositório
# ----------------------------------------------------
g_gramatica_index_repository = GGramaticaIndexRepository()
# ----------------------------------------------------
# Execução do repositório
# ----------------------------------------------------
response = g_gramatica_index_repository.execute()
# ----------------------------------------------------
# Retorno da informação
# --------------------------------
return response

View file

@ -1,38 +0,0 @@
from abstracts.action import BaseAction
from packages.v1.administrativo.repositories.g_gramatica.g_gramatica_save_repository import (
GGramaticaSaveRepository,
)
from packages.v1.administrativo.schemas.g_gramatica_schema import GGramaticaSaveSchema
class GGramaticaSaveAction(BaseAction):
"""
Serviço responsável por encapsular a lógica de negócio para a operação
de salvamento de um novo registro na tabela G_GRAMATICA.
"""
def execute(self, g_gramatica_save_schema: GGramaticaSaveSchema):
"""
Executa a operação de salvamento.
Args:
g_gramatica_schema (GGramaticaSchema):
O esquema com os dados a serem persistidos.
Returns:
O resultado da operação de salvamento.
"""
# ----------------------------------------------------
# Instanciamento do repositório
# ----------------------------------------------------
g_gramatica_save_repository = GGramaticaSaveRepository()
# ----------------------------------------------------
# Execução do repositório
# ----------------------------------------------------
response = g_gramatica_save_repository.execute(g_gramatica_save_schema)
# ----------------------------------------------------
# Retorno da informação
# ----------------------------------------------------
return response

View file

@ -1,38 +0,0 @@
from abstracts.action import BaseAction
from packages.v1.administrativo.repositories.g_gramatica.g_gramatica_show_repository import (
GGramaticaShowRepository,
)
from packages.v1.administrativo.schemas.g_gramatica_schema import GGramaticaIdSchema
class GGramaticaShowAction(BaseAction):
"""
Serviço responsável por encapsular a lógica de negócio para a exibição
de um registro na tabela G_NATUREZA_TITULO.
"""
def execute(self, g_gramatica_id_schema: GGramaticaIdSchema):
"""
Executa a operação de exibição.
Args:
g_gramatica_id_schema (GGramaticaIdSchema):
O esquema com o ID do registro a ser exibido.
Returns:
O resultado da operação de exibição.
"""
# ----------------------------------------------------
# Instanciamento do repositório
# ----------------------------------------------------
g_gramatica_show_repository = GGramaticaShowRepository()
# ----------------------------------------------------
# Execução do repositório
# ----------------------------------------------------
response = g_gramatica_show_repository.execute(g_gramatica_id_schema)
# ----------------------------------------------------
# Retorno da informação
# ----------------------------------------------------
return response

View file

@ -1,38 +0,0 @@
from abstracts.action import BaseAction
from packages.v1.administrativo.repositories.g_gramatica.g_gramatica_update_repository import (
GGramaticaUpdateRepository,
)
from packages.v1.administrativo.schemas.g_gramatica_schema import GGramaticaUpdateSchema
class GGramaticaUpdateAction(BaseAction):
"""
Serviço responsável por encapsular a lógica de negócio para a atualização
de um registro na tabela G_NATUREZA_TITULO.
"""
def execute(self, g_gramatica_update_schema: GGramaticaUpdateSchema):
"""
Executa a operação de atualização.
Args:
g_gramatica_update_schema (GGramaticaUpdateSchema):
O esquema com os dados a serem atualizados.
Returns:
O resultado da operação de atualização.
"""
# ----------------------------------------------------
# Instanciamento do repositório de atualização
# ----------------------------------------------------
g_gramatica_update_repository = GGramaticaUpdateRepository()
# ----------------------------------------------------
# Execução do repositório
# ----------------------------------------------------
response = g_gramatica_update_repository.execute(g_gramatica_update_schema)
# ----------------------------------------------------
# Retorno do resultado
# ----------------------------------------------------
return response

View file

@ -1,9 +1,11 @@
from abstracts.action import BaseAction from abstracts.action import BaseAction
# O repositório TServicoTipoDeleteRepository deve ser substituído pelo GMarcacaoTipoDeleteRepository.
from packages.v1.administrativo.repositories.g_marcacao_tipo.g_marcacao_tipo_delete_repository import \
DeleteRepository
# O schema TServicoTipoIdSchema deve ser substituído por GMarcacaoTipoIdSchema # O schema TServicoTipoIdSchema deve ser substituído por GMarcacaoTipoIdSchema
# que contém o campo-chave MARCACAO_TIPO_ID. # que contém o campo-chave MARCACAO_TIPO_ID.
from packages.v1.administrativo.schemas.g_marcacao_tipo_schema import GMarcacaoTipoIdSchema from packages.v1.administrativo.schemas.g_marcacao_tipo_schema import \
# O repositório TServicoTipoDeleteRepository deve ser substituído pelo GMarcacaoTipoDeleteRepository. GMarcacaoTipoIdSchema
from packages.v1.administrativo.repositories.g_marcacao_tipo.g_marcacao_tipo_delete_repository import DeleteRepository
class DeleteAction(BaseAction): class DeleteAction(BaseAction):

View file

@ -1,8 +1,10 @@
from abstracts.action import BaseAction from abstracts.action import BaseAction
# Ajuste do schema de entrada
from packages.v1.administrativo.schemas.g_marcacao_tipo_schema import GMarcacaoTipoDescricaoSchema
# Ajuste do repositório # Ajuste do repositório
from packages.v1.administrativo.repositories.g_marcacao_tipo.g_marcacao_tipo_get_by_descricao_repository import GetByDescricaoRepository from packages.v1.administrativo.repositories.g_marcacao_tipo.g_marcacao_tipo_get_by_descricao_repository import \
GetByDescricaoRepository
# Ajuste do schema de entrada
from packages.v1.administrativo.schemas.g_marcacao_tipo_schema import \
GMarcacaoTipoDescricaoSchema
class GetByDescricaoAction(BaseAction): class GetByDescricaoAction(BaseAction):

View file

@ -1,31 +0,0 @@
from abstracts.action import BaseAction
# Ajuste do schema de entrada
from packages.v1.administrativo.schemas.g_marcacao_tipo_schema import GMarcacaoTipoGrupoSchema
# Ajuste do repositório
from packages.v1.administrativo.repositories.g_marcacao_tipo.g_marcacao_tipo_get_by_grupo_repository import GetByGrupoRepository
class GetByGrupoAction(BaseAction):
"""
Serviço responsável por encapsular a lógica de negócio para a operação
de busca de um registro na tabela G_MARCACAO_TIPO por filtro.
"""
def execute(self, marcacao_tipo_schema: GMarcacaoTipoGrupoSchema):
"""
Executa a operação de busca no banco de dados.
Args:
marcacao_tipo_schema (GMarcacaoTipoGrupoSchema): O esquema com os filtros a serem buscados.
Returns:
O registro encontrado ou None.
"""
# Instanciamento do repositório
show_repository = GetByGrupoRepository()
# Execução do repositório
response = show_repository.execute(marcacao_tipo_schema)
# Retorno da informação
return response

View file

@ -1,6 +1,8 @@
from abstracts.action import BaseAction from abstracts.action import BaseAction
# O repositório TServicoTipoIndexRepository deve ser substituído pelo GMarcacaoTipoIndexRepository. # O repositório TServicoTipoIndexRepository deve ser substituído pelo GMarcacaoTipoIndexRepository.
from packages.v1.administrativo.repositories.g_marcacao_tipo.g_marcacao_tipo_index_repository import IndexRepository from packages.v1.administrativo.repositories.g_marcacao_tipo.g_marcacao_tipo_index_repository import \
IndexRepository
class IndexAction(BaseAction): class IndexAction(BaseAction):
""" """

View file

@ -1,8 +1,11 @@
from abstracts.action import BaseAction from abstracts.action import BaseAction
# Ajuste do schema de entrada
from packages.v1.administrativo.schemas.g_marcacao_tipo_schema import GMarcacaoTipoSaveSchema
# Ajuste do repositório # Ajuste do repositório
from packages.v1.administrativo.repositories.g_marcacao_tipo.g_marcacao_tipo_save_repository import SaveRepository from packages.v1.administrativo.repositories.g_marcacao_tipo.g_marcacao_tipo_save_repository import \
SaveRepository
# Ajuste do schema de entrada
from packages.v1.administrativo.schemas.g_marcacao_tipo_schema import \
GMarcacaoTipoSaveSchema
class SaveAction(BaseAction): class SaveAction(BaseAction):
""" """

View file

@ -1,8 +1,11 @@
from abstracts.action import BaseAction from abstracts.action import BaseAction
# Ajuste do schema de entrada
from packages.v1.administrativo.schemas.g_marcacao_tipo_schema import GMarcacaoTipoIdSchema
# Ajuste do repositório # Ajuste do repositório
from packages.v1.administrativo.repositories.g_marcacao_tipo.g_marcacao_tipo_show_repository import ShowRepository from packages.v1.administrativo.repositories.g_marcacao_tipo.g_marcacao_tipo_show_repository import \
ShowRepository
# Ajuste do schema de entrada
from packages.v1.administrativo.schemas.g_marcacao_tipo_schema import \
GMarcacaoTipoIdSchema
class ShowAction(BaseAction): class ShowAction(BaseAction):
""" """

View file

@ -1,16 +1,22 @@
# Importação do Schema ajustada # Importação do Schema ajustada
from packages.v1.administrativo.schemas.g_marcacao_tipo_schema import GMarcacaoTipoUpdateSchema
# Importação do Repositório ajustada # Importação do Repositório ajustada
from packages.v1.administrativo.repositories.g_marcacao_tipo.g_marcacao_tipo_update_repository import UpdateRepository from packages.v1.administrativo.repositories.g_marcacao_tipo.g_marcacao_tipo_update_repository import \
UpdateRepository
from packages.v1.administrativo.schemas.g_marcacao_tipo_schema import \
GMarcacaoTipoUpdateSchema
# A classe UpdateAction não herda de BaseAction no arquivo original, mantemos o padrão. # A classe UpdateAction não herda de BaseAction no arquivo original, mantemos o padrão.
class UpdateAction: class UpdateAction:
""" """
Service responsável por encapsular a lógica de negócio para a atualização Service responsável por encapsular a lógica de negócio para a atualização
de um registro na tabela G_MARCACAO_TIPO. de um registro na tabela G_MARCACAO_TIPO.
""" """
def execute(self, marcacao_tipo_id: int, marcacao_tipo_schema: GMarcacaoTipoUpdateSchema): def execute(
self, marcacao_tipo_id: int, marcacao_tipo_schema: GMarcacaoTipoUpdateSchema
):
""" """
Executa a operação de atualização. Executa a operação de atualização.

View file

@ -1,6 +1,8 @@
from abstracts.action import BaseAction from abstracts.action import BaseAction
from packages.v1.administrativo.schemas.g_medida_tipo_schema import GMedidaTipoIdSchema from packages.v1.administrativo.repositories.g_medida_tipo.g_medida_tipo_delete_repository import \
from packages.v1.administrativo.repositories.g_medida_tipo.g_medida_tipo_delete_repository import DeleteRepository DeleteRepository
from packages.v1.administrativo.schemas.g_medida_tipo_schema import \
GMedidaTipoIdSchema
class DeleteAction(BaseAction): class DeleteAction(BaseAction):

View file

@ -1,6 +1,8 @@
from abstracts.action import BaseAction from abstracts.action import BaseAction
from packages.v1.administrativo.schemas.g_medida_tipo_schema import GMedidaTipoDescricaoSchema from packages.v1.administrativo.repositories.g_medida_tipo.g_medida_tipo_get_by_descricao_repository import \
from packages.v1.administrativo.repositories.g_medida_tipo.g_medida_tipo_get_by_descricao_repository import GetByDescricaoRepository GetByDescricaoRepository
from packages.v1.administrativo.schemas.g_medida_tipo_schema import \
GMedidaTipoDescricaoSchema
class GetByDescricaoAction(BaseAction): class GetByDescricaoAction(BaseAction):

View file

@ -1,5 +1,7 @@
from abstracts.action import BaseAction from abstracts.action import BaseAction
from packages.v1.administrativo.repositories.g_medida_tipo.g_medida_tipo_index_repository import IndexRepository from packages.v1.administrativo.repositories.g_medida_tipo.g_medida_tipo_index_repository import \
IndexRepository
class IndexAction(BaseAction): class IndexAction(BaseAction):
""" """

View file

@ -1,6 +1,9 @@
from abstracts.action import BaseAction from abstracts.action import BaseAction
from packages.v1.administrativo.schemas.g_medida_tipo_schema import GMedidaTipoSaveSchema from packages.v1.administrativo.repositories.g_medida_tipo.g_medida_tipo_save_repository import \
from packages.v1.administrativo.repositories.g_medida_tipo.g_medida_tipo_save_repository import SaveRepository SaveRepository
from packages.v1.administrativo.schemas.g_medida_tipo_schema import \
GMedidaTipoSaveSchema
class SaveAction(BaseAction): class SaveAction(BaseAction):
""" """

View file

@ -1,6 +1,9 @@
from abstracts.action import BaseAction from abstracts.action import BaseAction
from packages.v1.administrativo.schemas.g_medida_tipo_schema import GMedidaTipoIdSchema from packages.v1.administrativo.repositories.g_medida_tipo.g_medida_tipo_show_repository import \
from packages.v1.administrativo.repositories.g_medida_tipo.g_medida_tipo_show_repository import ShowRepository ShowRepository
from packages.v1.administrativo.schemas.g_medida_tipo_schema import \
GMedidaTipoIdSchema
class ShowAction(BaseAction): class ShowAction(BaseAction):
""" """

View file

@ -1,11 +1,13 @@
from packages.v1.administrativo.schemas.g_medida_tipo_schema import GMedidaTipoUpdateSchema from packages.v1.administrativo.repositories.g_medida_tipo.g_medida_tipo_update_repository import \
from packages.v1.administrativo.repositories.g_medida_tipo.g_medida_tipo_update_repository import UpdateRepository UpdateRepository
from packages.v1.administrativo.schemas.g_medida_tipo_schema import \
GMedidaTipoUpdateSchema
class UpdateAction: class UpdateAction:
""" """
Service responsável por encapsular a lógica de negócio para a atualização Service responsável por encapsular a lógica de negócio para a atualização
de um registro na tabela g_medida_tipo. de um registro na tabela g_medida_tipo.
""" """
def execute(self, medida_tipo_id: int, medida_tipo_schema: GMedidaTipoUpdateSchema): def execute(self, medida_tipo_id: int, medida_tipo_schema: GMedidaTipoUpdateSchema):

View file

@ -1,6 +1,8 @@
from abstracts.action import BaseAction from abstracts.action import BaseAction
from packages.v1.administrativo.schemas.g_natureza_schema import GNaturezaIdSchema from packages.v1.administrativo.repositories.g_natureza.g_natureza_delete_repository import \
from packages.v1.administrativo.repositories.g_natureza.g_natureza_delete_repository import DeleteRepository DeleteRepository
from packages.v1.administrativo.schemas.g_natureza_schema import \
GNaturezaIdSchema
class DeleteAction(BaseAction): class DeleteAction(BaseAction):

View file

@ -1,6 +1,8 @@
from abstracts.action import BaseAction from abstracts.action import BaseAction
from packages.v1.administrativo.schemas.g_natureza_schema import GNaturezaDescricaoSchema from packages.v1.administrativo.repositories.g_natureza.g_natureza_get_by_descricao_repository import \
from packages.v1.administrativo.repositories.g_natureza.g_natureza_get_by_descricao_repository import GetByDescricaoRepository GetByDescricaoRepository
from packages.v1.administrativo.schemas.g_natureza_schema import \
GNaturezaDescricaoSchema
class GetByDescricaoAction(BaseAction): class GetByDescricaoAction(BaseAction):

View file

@ -1,5 +1,7 @@
from abstracts.action import BaseAction from abstracts.action import BaseAction
from packages.v1.administrativo.repositories.g_natureza.g_natureza_index_repository import IndexRepository from packages.v1.administrativo.repositories.g_natureza.g_natureza_index_repository import \
IndexRepository
class IndexAction(BaseAction): class IndexAction(BaseAction):
""" """

View file

@ -1,30 +0,0 @@
from abstracts.action import BaseAction
from packages.v1.administrativo.repositories.g_natureza.g_natureza_index_by_sistema_id_repository import (
IndexBySistemaIdRepository,
)
from packages.v1.administrativo.schemas.g_natureza_schema import (
GNaturezaSistemaIdSchema,
)
class IndexActionBySistemaId(BaseAction):
"""
Serviço responsável por encapsular a lógica de negócio para a operação
de listagem de todos os registros na tabela g_natureza.
"""
def execute(self, g_natureza_sistema_id_schema: GNaturezaSistemaIdSchema):
"""
Executa a operação de listagem no banco de dados.
Returns:
A lista de todos os registros.
"""
# Instanciamento do repositório
index_repository = IndexBySistemaIdRepository()
# Execução do repositório
response = index_repository.execute(g_natureza_sistema_id_schema)
# Retorno da informação
return response

View file

@ -1,6 +1,9 @@
from abstracts.action import BaseAction from abstracts.action import BaseAction
from packages.v1.administrativo.schemas.g_natureza_schema import GNaturezaSaveSchema from packages.v1.administrativo.repositories.g_natureza.g_natureza_save_repository import \
from packages.v1.administrativo.repositories.g_natureza.g_natureza_save_repository import SaveRepository SaveRepository
from packages.v1.administrativo.schemas.g_natureza_schema import \
GNaturezaSaveSchema
class SaveAction(BaseAction): class SaveAction(BaseAction):
""" """

Some files were not shown because too many files have changed in this diff Show more