Desarrollo de aplicaciones con un framework ORM de Python - Lindorm

Conéctese a LindormTable mediante SQLAlchemy y realice operaciones CRUD a través de su interfaz compatible con MySQL. Esta guía explica cómo instalar las dependencias, configurar la conexión, definir un modelo y ejecutar operaciones de inserción, consulta, actualización y eliminación.

Requisitos previos

Antes de empezar, asegúrese de contar con lo siguiente:

Python 3.7 o una versión posterior instalada
LindormTable versión 2.6.1 o posterior. Consulte Release notes of LindormTable y Upgrade the minor engine version of a Lindorm instance
La compatibilidad con MySQL habilitada en su instancia. Consulte Enable the MySQL compatibility feature
La dirección IP de su cliente agregada a la lista blanca de la instancia. Consulte Configure a whitelist

Instalación de dependencias

LindormTable expone un endpoint compatible con MySQL. Para conectarse desde Python se requieren dos bibliotecas:

SQLAlchemy: capa ORM que permite asignar clases de Python a tablas de base de datos y escribir consultas en Python en lugar de SQL sin procesar
PyMySQL: controlador de base de datos que gestiona la comunicación subyacente del protocolo MySQL; SQLAlchemy no incluye un controlador

Instale ambas bibliotecas:

pip install PyMySQL
pip install SQLAlchemy

Configuración de la conexión

Utilice create_engine() para inicializar la conexión a su instancia de LindormTable.

from sqlalchemy import create_engine

engine = create_engine(
    'mysql+pymysql://<user>:<password>@<lindorm_mysql_addr>:33060/<database>',
    echo=True
)

Parámetro	Descripción
`<user>`	Su nombre de usuario de LindormTable. Para restablecer una contraseña olvidada, consulte Modify a user password
`<password>`	La contraseña del usuario. Para restablecer una contraseña olvidada, consulte Modify a user password
`<lindorm_mysql_addr>`	El endpoint compatible con MySQL. Consulte View endpoints
`<database>`	El nombre de la base de datos de destino. El valor predeterminado es `default`

Selección del tipo de endpoint:

Despliegue	Endpoint recomendado	Cómo obtenerlo
Aplicación en una instancia ECS (recomendado)	Endpoint VPC compatible con MySQL: ofrece mayor seguridad y menor latencia de red	Consulte View endpoints
Aplicación desplegada en instalaciones locales	Endpoint de Internet compatible con MySQL	Antes de conectarse, habilite el endpoint público en la consola de Lindorm: vaya a Database Connections > Wide Table Engine y haga clic en Enable Public Endpoint

Definición de un modelo

Defina la estructura de su tabla como una clase de Python mediante la base declarativa de SQLAlchemy.

from sqlalchemy import Column, String, Integer, Float
from sqlalchemy.orm import declarative_base

Base = declarative_base()

class Player(Base):
    __tablename__ = 'player'
    player_id = Column(Integer, primary_key=True, autoincrement=False)
    player_name = Column(String(255))
    player_height = Column(Float)

    def __str__(self):
        return " ".join(str(item) for item in (self.player_id, self.player_name, self.player_height))

Ejecución de operaciones CRUD

Todos los ejemplos utilizan una sesión de SQLAlchemy para interactuar con LindormTable.

Creación de la tabla e inserción de datos

from sqlalchemy.orm import sessionmaker

# Create the table based on the model definition
Base.metadata.create_all(engine)

Session = sessionmaker(bind=engine)
session = Session()

# Insert a single row
new_player = Player(player_id=1001, player_name="john", player_height=2.08)
session.add(new_player)

# Insert multiple rows
session.add_all([
    Player(player_id=1002, player_name="bob", player_height=1.65),
    Player(player_id=1003, player_name="smith", player_height=1.82),
])
session.commit()

Consulta de datos

# Query rows matching a filter condition
rows = session.query(Player).filter(Player.player_id == 1001).all()
print([row.__str__() for row in rows])

Actualización de datos

# Update a field for rows matching a filter condition
session.query(Player).filter(Player.player_id == 1003).update({"player_name": "brown"})
session.commit()

Eliminación de datos

# Delete rows matching a filter condition
session.query(Player).filter(Player.player_id == 1002).delete()
session.commit()

Código de ejemplo completo

A continuación se muestra un ejemplo funcional completo. Reemplace los marcadores de posición de la cadena de conexión con sus valores reales antes de ejecutarlo.

from sqlalchemy import create_engine
from sqlalchemy import Column, String, Integer, Float
from sqlalchemy.orm import declarative_base
from sqlalchemy.orm import sessionmaker

Base = declarative_base()

# Define the Player model
class Player(Base):
    __tablename__ = 'player'
    player_id = Column(Integer, primary_key=True, autoincrement=False)
    player_name = Column(String(255))
    player_height = Column(Float)

    def __str__(self):
        return " ".join(str(item) for item in (self.player_id, self.player_name, self.player_height))

# Initialize the connection. Replace the placeholders with your actual values.
engine = create_engine(
    'mysql+pymysql://user:test@ld-bp1l39h99192d****-proxy-sql-lindorm-public.lindorm.rds.aliyuncs.com:33060/default',
    echo=True
)
Session = sessionmaker(bind=engine)

# Create the table
Base.metadata.create_all(engine)

# Open a session
session = Session()

# Insert a single row
new_player = Player(player_id=1001, player_name="john", player_height=2.08)
session.add(new_player)

# Insert multiple rows
session.add_all([
    Player(player_id=1002, player_name="bob", player_height=1.65),
    Player(player_id=1003, player_name="smith", player_height=1.82),
])
session.commit()

# Query rows where player_id is 1001
rows = session.query(Player).filter(Player.player_id == 1001).all()
print([row.__str__() for row in rows])

# Delete the row where player_id is 1002
session.query(Player).filter(Player.player_id == 1002).delete()
session.commit()

# Update the row where player_id is 1003
session.query(Player).filter(Player.player_id == 1003).update({"player_name": "brown"})
session.commit()

# Query all remaining rows
rows = session.query(Player).all()
print([row.__str__() for row in rows])

Salida esperada:

['1001 john 2.08']
['1001 john 2.08', '1003 brown 1.82']

Pasos siguientes

View endpoints — obtenga las direcciones de los endpoints VPC y público de su instancia
Configure a whitelist — controle qué direcciones IP pueden conectarse a su instancia
Enable the MySQL compatibility feature — habilite la compatibilidad con el protocolo MySQL si aún no lo ha hecho