Documentação do projeto:

Bases: BaseModel

Modelo de dados para as vendas.

Parameters:

Name	Type	Description	Default
email	EmailStr	email do comprador	required
data	datetime	data da compra	required
valor	PositiveFloat	valor da compra	required
quantidade	PositiveInt	quantidade de produtos	required
produto	Produto	categoria do produto	required

Source code in src/contract.py

class Vendas(BaseModel):
    """
    Modelo de dados para as vendas.

    Args:
        email (EmailStr): email do comprador
        data (datetime): data da compra
        valor (PositiveFloat): valor da compra
        quantidade (PositiveInt): quantidade de produtos
        produto (Produto): categoria do produto
    """

    email: EmailStr
    data: datetime
    valor: PositiveFloat
    quantidade: PositiveInt
    produto: Produto

Salva os dados validados no PostgreSQL.

Source code in src/database.py

def salvar_no_postgres(dados: Vendas):
    """
    Salva os dados validados no PostgreSQL.
    """
    try:
        # Conecta ao banco de dados
        conn = psycopg2.connect(
            host=DB_HOST, database=DB_NAME, user=DB_USER, password=DB_PASS
        )
        cursor = conn.cursor()

        # Insere os dados na tabela de vendas
        insert_query = sql.SQL(
            "INSERT INTO vendas (email, data, valor, quantidade, produto) VALUES (%s, %s, %s, %s, %s)"
        )
        cursor.execute(
            insert_query,
            (
                dados.email,
                dados.data,
                dados.valor,
                dados.quantidade,
                dados.produto.value,
            ),
        )
        conn.commit()
        cursor.close()
        conn.close()
        print("Dados salvos com sucesso no banco de dados!")
    except Exception as e:
        raise Exception(f"Erro ao salvar no banco de dados: {e}")

Cria a tabela de vendas no PostgreSQL.

Source code in src/database.py

def criar_tabela_vendas():
    """
    Cria a tabela de vendas no PostgreSQL.
    """
    try:
        # Conecta ao banco de dados
        conn = psycopg2.connect(
            host=DB_HOST, database=DB_NAME, user=DB_USER, password=DB_PASS
        )
        cursor = conn.cursor()

        # Cria a tabela de vendas
        create_table_query = sql.SQL(
            """
            CREATE TABLE IF NOT EXISTS vendas (
                id SERIAL PRIMARY KEY,
                email VARCHAR(255) NOT NULL,
                data TIMESTAMP WITHOUT TIME ZONE NOT NULL,
                valor NUMERIC(10, 2) NOT NULL,
                quantidade INTEGER NOT NULL,
                produto VARCHAR(255) NOT NULL
            )
            """
        )
        cursor.execute(create_table_query)
        conn.commit()
        cursor.close()
        conn.close()
    except Exception as e:
        raise Exception(f"Erro ao criar a tabela de vendas: {e}")

Salva uma lista de objetos Vendas no PostgreSQL em um único lote.

Source code in src/database.py

def salvar_no_postgres_em_lote(vendas: list[Vendas]):
    """
    Salva uma lista de objetos Vendas no PostgreSQL em um único lote.
    """
    try:
        # Conecta ao banco de dados
        conn = psycopg2.connect(
            host=DB_HOST, database=DB_NAME, user=DB_USER, password=DB_PASS
        )
        cursor = conn.cursor()

        # Prepara os valores para inserção em lote
        values = [
            (
                venda.email,
                venda.data,
                venda.valor,
                venda.quantidade,
                venda.produto.value,
            )
            for venda in vendas
        ]

        # Insere os dados na tabela de vendas em lote
        insert_query = sql.SQL(
            "INSERT INTO vendas (email, data, valor, quantidade, produto) VALUES %s"
        )
        execute_values(cursor, insert_query, values)
        conn.commit()
        cursor.close()
        conn.close()

        return True
    except Exception as e:
        raise Exception(f"Erro ao salvar no banco de dados: {e}")
        return False

Deleta todos os dados da tabela de vendas no PostgreSQL.

Source code in src/database.py

def delete_all_sales_data():
    """
    Deleta todos os dados da tabela de vendas no PostgreSQL.
    """
    try:
        # Conecta ao banco de dados
        conn = psycopg2.connect(
            host=DB_HOST, database=DB_NAME, user=DB_USER, password=DB_PASS
        )
        cursor = conn.cursor()

        # Deleta todos os dados da tabela de vendas
        delete_query = sql.SQL("DELETE FROM vendas")
        cursor.execute(delete_query)
        conn.commit()
        cursor.close()
        conn.close()

        return True
    except Exception as e:
        return False, f"Erro ao deletar os dados do banco de dados: {e}"