GGontijo

import json
import logging
import time
from typing import Callable

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

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

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

#!/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

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

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

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

# 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

from datetime import time
import logging
import asyncio
from fastapi import FastAPI, Response
from fastapi.concurrency import asynccontextmanager
from pydantic import BaseModel
from pygen import anexar_arquivo, baixar_planilha, executar_atividade
from guias import Imposto, emitir_guia_difal, emitir_guia_fethab, emitir_guia_iagro
from apuracao import processar_apuracao_difal, processar_apuracao_fethab, processar_apuracao_iagro
import urllib3

GGontijo

import logging
from time import sleep

class Threadlock():
    '''Singleton'''
    _instance = None
    _locked = False

    def is_locked(self):
        return self._locked