ORM : SQLAlchemy

Pots consultar més informació sobre SQLAlchemy en la seva documentació oficial.

Instal·lació de SQLAlchemy

Per instal·lar SQLAlchemy podem utilitzar el pip. A més, necessitem el driver de PostgreSQL:

pip install sqlalchemy psycopg2

Exemple: Creació de taules

El primer que es fa amb un ORM es definir el model de dades. Cada classe del nostre ORM es transformarà en una taula, i cada atribut en un camp.

Creem dintre de postgresql una base de dades que l'anomenarem db_agenda

CREATE DATABASE db_agenda;
CREATE USER uagenda WITH PASSWORD '12345';
GRANT ALL PRIVILEGES ON DATABASE db_agenda TO uagenda;

Ara escriurem el codi utilitzant el ORM per tal de que ens crei les taules.

from sqlalchemy import create_engine, Column, Integer, String, Date, ForeignKey
from sqlalchemy.orm import DeclarativeBase

engine = create_engine('postgresql+psycopg2://uagenda:12345@localhost:5432/db_agenda')

class Base(DeclarativeBase):
    pass

class Persona(Base):
    __tablename__ = 'persona'
    id_p = Column(Integer, primary_key=True, autoincrement=True)
    nom = Column(String)
    cognoms = Column(String)
    data_naixement = Column(Date)

class Telefon(Base):
    __tablename__ = 'telefon'
    id_t = Column(Integer, primary_key=True, autoincrement=True)
    persona_id = Column(Integer, ForeignKey('persona.id_p'))
    numero = Column(String(20))

Base.metadata.create_all(engine)

Si anem a mirar la base de dades després d'executar el programa veurem que ens ha creat les taules.

Es poden consultar quins tipus de dades podem tenir en el model, en la documentació de tipus de SQLAlchemy.

Exemple: Inserció de registres

from sqlalchemy import create_engine, Column, Integer, String, Date, ForeignKey
from sqlalchemy.orm import DeclarativeBase, Session

engine = create_engine('postgresql+psycopg2://uagenda:12345@localhost:5432/db_agenda')

class Base(DeclarativeBase):
    pass

class Persona(Base):
    __tablename__ = 'persona'
    id_p = Column(Integer, primary_key=True, autoincrement=True)
    nom = Column(String)
    cognoms = Column(String)
    data_naixement = Column(Date)

class Telefon(Base):
    __tablename__ = 'telefon'
    id_t = Column(Integer, primary_key=True, autoincrement=True)
    persona_id = Column(Integer, ForeignKey('persona.id_p'))
    numero = Column(String(20))

with Session(engine) as sessio:
    agarcia = Persona(nom='antoni', cognoms='garcia sanchez', data_naixement='1978-09-21')
    sessio.add(agarcia)
    sessio.flush()  # per obtenir l'id_p generat

    tel_agarcia = Telefon(persona_id=agarcia.id_p, numero='972335794')
    sessio.add(tel_agarcia)
    sessio.commit()

Exemple: Modificació de registres

from sqlalchemy import create_engine, Column, Integer, String, Date, ForeignKey, update
from sqlalchemy.orm import DeclarativeBase, Session

engine = create_engine('postgresql+psycopg2://uagenda:12345@localhost:5432/db_agenda')

class Base(DeclarativeBase):
    pass

class Persona(Base):
    __tablename__ = 'persona'
    id_p = Column(Integer, primary_key=True, autoincrement=True)
    nom = Column(String)
    cognoms = Column(String)
    data_naixement = Column(Date)

class Telefon(Base):
    __tablename__ = 'telefon'
    id_t = Column(Integer, primary_key=True, autoincrement=True)
    persona_id = Column(Integer, ForeignKey('persona.id_p'))
    numero = Column(String(20))

with Session(engine) as sessio:
    sessio.execute(
        update(Telefon).where(Telefon.id_t == 1).values(numero='972356758')
    )
    sessio.commit()

Exemple: Eliminar registres

from sqlalchemy import create_engine, Column, Integer, String, Date, ForeignKey, delete
from sqlalchemy.orm import DeclarativeBase, Session

engine = create_engine('postgresql+psycopg2://uagenda:12345@localhost:5432/db_agenda')

class Base(DeclarativeBase):
    pass

class Persona(Base):
    __tablename__ = 'persona'
    id_p = Column(Integer, primary_key=True, autoincrement=True)
    nom = Column(String)
    cognoms = Column(String)
    data_naixement = Column(Date)

class Telefon(Base):
    __tablename__ = 'telefon'
    id_t = Column(Integer, primary_key=True, autoincrement=True)
    persona_id = Column(Integer, ForeignKey('persona.id_p'))
    numero = Column(String(20))

with Session(engine) as sessio:
    sessio.execute(
        delete(Telefon).where(Telefon.numero == '972356758')
    )
    sessio.commit()

Exemple: Consulta de registres

from sqlalchemy import create_engine, Column, Integer, String, Date, ForeignKey, select
from sqlalchemy.orm import DeclarativeBase, Session

engine = create_engine('postgresql+psycopg2://uagenda:12345@localhost:5432/db_agenda')

class Base(DeclarativeBase):
    pass

class Persona(Base):
    __tablename__ = 'persona'
    id_p = Column(Integer, primary_key=True, autoincrement=True)
    nom = Column(String)
    cognoms = Column(String)
    data_naixement = Column(Date)

class Telefon(Base):
    __tablename__ = 'telefon'
    id_t = Column(Integer, primary_key=True, autoincrement=True)
    persona_id = Column(Integer, ForeignKey('persona.id_p'))
    numero = Column(String(20))

with Session(engine) as sessio:
    consulta = sessio.execute(select(Persona)).scalars().all()
    for p in consulta:
        print(f"{p.cognoms}, {p.nom}")