Consulta Select en SQLAlchemy

Seguimos con la serie de conectarnos a una base de datos mediante SQLAlchemy, en este caso utilizare variables de entorno almacenadas en un archivo .env para facilitar la distribución de nuestro programa y mejorar un poco la seguridad de nuestro servidor de base de datos.

Una excelente practica es modularizar nuestro código dejando un modulo para la conexión de la base de datos, otro modulo con los modelos o clases de la base de datos y las queries.

Iniciaremos creando el modulo de la conexión de la base de datos, el cual nombrare server.py y contendrá el siguiente código:
import os

from dotenv import load_dotenv
from sqlalchemy import create_engine

lee las variables de entorno

load_dotenv()

def create_engine_data()->"create_engine":
ENGINE_DB = os.getenv('ENGINE_DB')
CONNECTOR = os.getenv('CONNECTOR')
USER_DB = os.getenv('USER_DB')
PSW = os.getenv('PSW')
IP_SERVER = os.getenv('IP_SERVER')
DATABASE = os.getenv('DATABASE')
try:
engine = create_engine(
f'{ENGINE_DB}+{CONNECTOR}://{USER_DB}:\
{PSW}@{IP_SERVER}/{DATABASE}')
except Exception as ex:
raise ex
return engine

Importamos las variables de entorno con el uso de os.getenv, el nombre de cada constante es como se enuentra en nuestro archivo .env, luego procedemos a tratar de crear una instancia del motor de base de datos en el caso que la conexión se logre retornamos la instancia y si existe algún tipo de problema al momento de realizar la conexión nos levantaría una excepción.

Nota: Si están trabajando con un programa un poco mas complejo, lo adecuado seria importar el dontev y cargar los dotenv al inicio del programa y no en el archivo server.py quedando las importaciones con dos líneas menos.

import os

from sqlalchemy import create_engine

Fin de las importaciones

La función se mantiene intacta

Con el modulo de conexión al servidor de base de datos creado, podemos crear el modulo o los modelos de la base de datos que necesitaremos. Es importante aclarar que no es necesario modelar toda la base de datos, si solo necesitamos dos o tres tablas para trabajar basta con modelar estas tablas, asimismo no es necesario modelar toda la tabla si vamos a trabajar con 2 o 3 campos podemos modelar solamente esos campos.

Nuestra tabla usuarios tiene los siguientes campos:

A modo de ejemplo dejare por fuera el campo telephone, solamente porque no lo necesitaría.

El código de nuestro modelo de la tabla seria como el siguiente:
from sqlalchemy import Column
from sqlalchemy import String
from sqlalchemy.orm import declarative_base

Base = declarative_base()

class Usuario(Base):
#Nombre de la tabla en la base de datos
tablename = "usuarios"

username = Column(String(30), primary_key=True)
name = Column(String(50))
surname = Column(String(50))
email = Column(String(60), unique=True)

#Representacion para el programador
def __repr__(self):
    return f"Usuario(username={self.username!r}, name={self.name!r},\
        surname={self.surname!r}, email={self.email!r})"

Con nuestro modelo hecho podemos continuar con el proceso de acceder a los datos en nuestra base de datos mediante el ORM, debemos crear la consulta o query con el ORM, en este caso hare dos consultas una que obtenga todos los usuarios y otra que obtenga los usuarios que cumplen con un criterio en especifico.

from sqlalchemy import select
from sqlalchemy.orm import Session

from server import create_engine_data
from models import Usuario

ENGINE = create_engine_data()
SESSION = Session(ENGINE)

def obtener_usuarios(name=''):
if name == '':
stmt = select(Usuario)
else:
stmt = select(Usuario).where(Usuario.name==name)
for usuario in SESSION.scalars(stmt):
print(usuario)

La consulta SQL para extraer todos lo usuarios seria
SELECT username, name, surname, telephone FROM usuarios

Y La segunda consulta SQL para extraer los usuarios que cumplen con un criterio seria
SELECT username, name, surname, telephone FROM usuario where name = name

Si deseamos ejecutar el código SQL de forma directa con ayuda de SQLAlchemy podemos hacerlo con ayuda de la instancia del servidor:

from server import create_engine_data
with ENGINE.connect() as conn:
data = conn.execute("SELECT username, name, surname, telephone FROM usuarios")

for usuario in data:
    print(usuario)

Nota: Debemos tener en cuenta que si migramos de motor de base de datos seria mas dificultoso el proceso de migración ya que deberíamos de verificar y reestructurar cada una de las sentencias.