GGontijo

# Se você não quiser mapear uma tabela do sqlalchemy e esta tendo erros por conta de colunas
# no pydantic que não existem em sua tabela, você pode filtra-las desta forma

# If you don't want to map an SQLAlchemy table and are getting errors due to columns
# in Pydantic that don't exist in your table, you can filter them this way

def filtrar_colunas_validas(data, table) -> dict:
    """Remove colunas que não são utilizadas na tabela"""
    valid_columns = set(table.columns.keys())

GGontijo

class Ipva(BaseModel):
    veiculo: Veiculo
    status: str = Field(alias='STATUS')
    ano: Optional[str] = Field(None, alias='ANO')
    valor: Optional[str] = Field(None, alias='VALOR')
    juros: Optional[str] = Field(None, alias='JUROS')
    multa: Optional[str] = Field(None, alias='MULTA')
    correcao: Optional[str] = Field(None, alias='CORRECAO')
    tse: Optional[str] = Field(None, alias='TSE')
    desconto: Optional[str] = Field(None, alias='DESCONTO')

GGontijo

version: "3.8"
services:
  rabbitmq:
    image: rabbitmq:4.0.3-management
    container_name: rabbitmq_rpa
    hostname: rmq
    restart: always
    command: >
      bash -c '
        rabbitmq-server &

GGontijo

import json
from pika import BlockingConnection, ConnectionParameters, PlainCredentials, BasicProperties
import uuid
import time

response = None
# Identificador único para a mensagem enviada
corr_id = str(uuid.uuid4())

def on_response(ch, method, properties, body):

GGontijo

#!/bin/bash

# Atualizar o cache de pacotes
sudo apt update -y

# Verificar se o Chrome está instalado
if ! command -v google-chrome &> /dev/null; then
    echo "Google Chrome não encontrado. Instalando..."
    wget -q -O google-chrome.deb https://dl.google.com/linux/direct/google-chrome-stable_current_amd64.deb
    sudo dpkg -i google-chrome.deb

GGontijo

from abc import ABC, abstractmethod
from enum import Enum
import json
from typing import Callable
import pika
import time
import logging
from pydantic import BaseModel
from contextlib import contextmanager

GGontijo

#https://www.geeksforgeeks.org/diagnosing-and-fixing-memory-leaks-in-python/
from contextlib import asynccontextmanager
import logging
import tracemalloc

from fastapi import FastAPI, Request

# Start Tracemalloc
@asynccontextmanager
async def lifespan(app: FastAPI):

GGontijo

import json
import logging
import time
from typing import Callable

import pika
from pika import (
    BasicProperties,
    ConnectionParameters,
    PlainCredentials,

GGontijo

import json
from datetime import datetime, timedelta
from itertools import combinations


def parse_date(s: str):
    # Organize costuma usar YYYY-MM-DD em "date"
    return datetime.strptime(s, "%Y-%m-%d").date()