Imagem de destaque do blog justcode.com.br.

Como utilizar o SQLAlchemy com o banco de dados PostgreSQL

Hoje veremos como utilizar a linguagem de programação Python e o SQLAlchemy com o banco de dados PostgreSQL através do conector psycopg.

SQLAlchemy

O SQLAlchemy é uma biblioteca em Python que fornece uma maneira de interagir com bancos de dados relacionais de forma flexível e poderosa.

Ele permite que você trabalhe com bancos de dados SQL de uma maneira orientada a objetos, mapeando tabelas de banco de dados para classes em Python e consultando essas tabelas usando uma sintaxe semelhante ao SQL.

Em resumo, o SQLAlchemy simplifica o acesso e a manipulação de bancos de dados SQL em aplicativos Python.

O SQLAlchemy pode ser instalado usando o pip, que é o gerenciador de pacotes padrão do Python:

pip install SQLAlchemy

Psycopg

Psycopg é um adaptador de banco de dados PostgreSQL para a linguagem de programação Python. Ele permite que os programas Python se conectem a um banco de dados PostgreSQL, enviem consultas SQL e manipulem os resultados dessas consultas.

O Psycopg é amplamente utilizado devido à sua facilidade de uso e eficiência, sendo uma escolha popular para desenvolvedores que trabalham com Python e PostgreSQL.


Posts relacionados:


    O psycopg pode ser instalado usando o pip, que é o gerenciador de pacotes padrão do Python:

    pip install "psycopg[binary]"

    Depois de instalado, você pode importar o conector em seus scripts Python e usá-lo para se conectar a um servidor PostgreSQL e executar operações de banco de dados.


    Contêiner

    Para testar o código de exemplo foi utilizando o seguinte arquivo docker-compose.yaml:

    services:
      db:
        image: postgres:16.2
        container_name: PostgreSQL
        restart: on-failure
        shm_size: 128mb
        ports:
          - '5432:5432'
        environment:
          POSTGRES_USER: dbuser
          POSTGRES_PASSWORD: 123456
          POSTGRES_DB: database_name
    

    📝 Contêiner testado com Docker e com o Podman.


    Exemplo de CRUD

    # -*- coding: utf-8 -*-
    """CRUD - SQLAlchemy - psycopg - PostgreSQL."""
    
    from sqlalchemy import (SmallInteger, String, URL,
                            create_engine, insert, select, delete, update)
    from sqlalchemy.orm import DeclarativeBase, Mapped, sessionmaker, mapped_column
    
    url_object = URL.create(
        drivername='postgresql+psycopg',
        username='dbuser',
        password='123456',
        host='localhost',
        port='5432',
        database='database_name',
    )
    
    engine = create_engine(url_object)
    
    Session = sessionmaker(bind=engine)
    
    
    class Base(DeclarativeBase):
        pass
    
    
    class TableName(Base):
        __tablename__ = 'table_name'
    
        id: Mapped[int] = mapped_column(primary_key=True)
        name: Mapped[str] = mapped_column('name', String(32))
        age: Mapped[int] = mapped_column('age', SmallInteger)
    
        def __repr__(self) -> str:
            return f'TableName(id={self.id}, name={self.name}, age={self.age})'
    
    
    if __name__ == '__main__':
        # Removing all tables from the database.
        Base.metadata.drop_all(engine)
    
        # Creating all tables.
        Base.metadata.create_all(engine)
    
        # Creating a session (add, commit, query, etc.).
        session = Session()
    
        # Create.
        print('[!] Create [!]')
        session.execute(
            insert(TableName)
            .values(
                name='renato',
                age=35,
            ),
        )
    
        # Returning the object that will be created.
        result = session.scalar(
            insert(TableName)
            .values(
                name='josé',
                age=29,
            )
            .returning(TableName),
        )
        session.commit()
        print(result)
    
        # Bulk create.
        session.execute(
            insert(TableName),
            [
                {'name': 'maria', 'age': 25},
                {'name': 'sandy', 'age': 19},
            ],
        )
    
        result = session.scalars(
            insert(TableName)
            .returning(TableName),
            [
                {'name': 'patrick', 'age': 33},
                {'name': 'gisele', 'age': 21},
            ],
        )
        session.commit()
        print(result.all())
    
        # Read.
        print('\n[!] Read [!]')
        result = session.scalars(
            select(TableName),
        )
        print(result.all())
    
        # Get by id.
        result = session.get(TableName, 1)
        print(result)
    
        # Limit.
        result = session.scalars(
            select(TableName)
            .limit(3),
        )
        print(result.all())
    
        # Where.
        result = session.scalars(
            select(TableName)
            .where(TableName.age > 30),
        )
        print(result.all())
    
        result = session.scalar(
            select(TableName)
            .where(TableName.id == 1),
        )
        print(result)
    
        # Filter.
        result = session.scalars(
            select(TableName)
            .filter_by(name='renato'),
        )
        print(result.all())
    
        # Update.
        print('\n[!] Update [!]')
        print(session.get(TableName, 1))
        session.execute(
            update(TableName)
            .where(TableName.id == 1)
            .values(name='joão'),
        )
        print(session.get(TableName, 1))
    
        print(session.get(TableName, 2))
        result = session.scalar(
            select(TableName)
            .where(TableName.id == 2),
        )
        result.name = 'antônio'
        session.commit()
        print(session.get(TableName, 2))
    
        # Delete.
        print('\n[!] Delete [!]')
        print(session.get(TableName, 1))
        session.execute(
            delete(TableName)
            .where(TableName.id == 1),
        )
        print(session.get(TableName, 1))
    
        print(session.get(TableName, 2))
        result = session.scalar(
            select(TableName)
            .where(TableName.id == 2),
        )
        session.delete(result)
        session.commit()
        print(session.get(TableName, 2))
    
        session.close()
    

    Conclusão

    O PostgreSQL é um sistema de gerenciamento de banco de dados poderoso e versátil, adequado para uma ampla gama de aplicações.

    Compreender os conceitos básicos, como bancos de dados, tabelas, consultas SQL e chaves, pode ajudá-lo a utilizar o PostgreSQL de forma eficaz em seus projetos de desenvolvimento.