Compartir comentarios
Las respuestas se generan en base a la documentación.

Crea el proyecto Python

Tabla de contenidos

Inicializa el proyecto

Comienza creando un proyecto Python con un entorno virtual:

$ mkdir tc-python-demo
$ cd tc-python-demo
$ python3 -m venv venv
$ source venv/bin/activate

Esta guía utiliza psycopg3 para interactuar con la base de datos Postgres, pytest para las pruebas y testcontainers-python para ejecutar una base de datos PostgreSQL en un contenedor.

Instala las dependencias:

$ pip install "psycopg[binary]" pytest testcontainers[postgres]
$ pip freeze > requirements.txt

El comando pip freeze genera un archivo requirements.txt para que otros puedan instalar las mismas versiones de los paquetes utilizando pip install -r requirements.txt.

Crea el helper de la base de datos

Crea un archivo db/connection.py con una función para obtener una conexión a la base de datos:

import os

import psycopg


def get_connection():
    host = os.getenv("DB_HOST", "localhost")
    port = os.getenv("DB_PORT", "5432")
    username = os.getenv("DB_USERNAME", "postgres")
    password = os.getenv("DB_PASSWORD", "postgres")
    database = os.getenv("DB_NAME", "postgres")
    return psycopg.connect(f"host={host} dbname={database} user={username} password={password} port={port}")

En lugar de escribir los parámetros de conexión a la base de datos directamente en el código (hard-code), la función utiliza variables de entorno. Esto hace posible ejecutar la aplicación en diferentes entornos sin cambiar el código.

Crea la lógica de negocio

Crea un archivo customers/customers.py y define la clase Customer:

class Customer:
    def __init__(self, cust_id, name, email):
        self.id = cust_id
        self.name = name
        self.email = email

    def __str__(self):
        return f"Customer({self.id}, {self.name}, {self.email})"

Añade una función create_table() para crear la tabla customers:

from db.connection import get_connection


def create_table():
    with get_connection() as conn:
        with conn.cursor() as cur:
            cur.execute("""
                CREATE TABLE customers (
                    id serial PRIMARY KEY,
                    name varchar not null,
                    email varchar not null unique)
                """)
            conn.commit()

La función obtiene una conexión a la base de datos mediante get_connection() y crea la tabla customers. La declaración with cierra automáticamente la conexión al finalizar.

Añade las funciones CRUD restantes:

def create_customer(name, email):
    with get_connection() as conn:
        with conn.cursor() as cur:
            cur.execute(
                "INSERT INTO customers (name, email) VALUES (%s, %s)", (name, email))
            conn.commit()


def get_all_customers() -> list[Customer]:
    with get_connection() as conn:
        with conn.cursor() as cur:
            cur.execute("SELECT * FROM customers")
            return [Customer(cid, name, email) for cid, name, email in cur]


def get_customer_by_email(email) -> Customer:
    with get_connection() as conn:
        with conn.cursor() as cur:
            cur.execute("SELECT id, name, email FROM customers WHERE email = %s", (email,))
            (cid, name, email) = cur.fetchone()
            return Customer(cid, name, email)


def delete_all_customers():
    with get_connection() as conn:
        with conn.cursor() as cur:
            cur.execute("DELETE FROM customers")
            conn.commit()
Note

Para mantener la guía sencilla, cada función crea una nueva conexión. En una aplicación real, utiliza un pool de conexiones para reutilizar las conexiones.

Escribe pruebas con Testcontainers »