Hoje veremos como utilizar a linguagem de programação Python e o SQLAlchemy com o banco de dados MySQL através do conector mysql-connector-python.
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 pip SQLAlchemy
Mysql-connector-python
O mysql-connector-python é o conector oficial do banco de dados MySQL para a linguagem de programação Python.
Ele permite que os desenvolvedores escrevam código Python para interagir com um banco de dados MySQL, executando consultas, inserindo dados, atualizando registros e muito mais.
Posts relacionados:
O mysql-connector-python
pode ser instalado usando o pip
, que é o gerenciador de pacotes padrão do Python:
pip install mysql-connector-python
Depois de instalado, você pode importar o conector em seus scripts Python e usá-lo para se conectar a um servidor MySQL 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: mysql:8.4.0
# image: mysql:5.7.44
container_name: MySQL
restart: on-failure
ports:
- '3306:3306'
environment:
MYSQL_USER: dbuser
MYSQL_PASSWORD: 123456
MYSQL_ROOT_PASSWORD: 123456
MYSQL_DATABASE: database_name
Exemplo de CRUD
# -*- coding: utf-8 -*-
'''CRUD - SQLAlchemy - MySQL Connector - MySQL.'''
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='mysql+mysqlconnector',
username='dbuser',
password='123456',
host='localhost',
port='3306',
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,
),
)
# MySQL does not support returning.
# 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},
],
)
# MySQL does not support returning.
# resutl = 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 MySQL é uma ferramenta poderosa para armazenar e gerenciar dados de forma eficiente.
Compreender os conceitos básicos, como bancos de dados, tabelas, consultas SQL e chaves, pode ajudá-lo a utilizar o MySQL de forma eficaz em seus projetos de desenvolvimento.