Lindorm:Develop applications based on a Python ORM framework

Prerequisites

Before you begin, make sure you have:

• Python 3.7 or later installed

• LindormTable version 2.6.1 or later. See Release notes of LindormTable and Upgrade the minor engine version of a Lindorm instance

• MySQL compatibility enabled on your instance. See Enable the MySQL compatibility feature

• Your client IP address added to the instance whitelist. See Configure a whitelist

Install dependencies

LindormTable exposes a MySQL-compatible endpoint. Connecting to it from Python requires two libraries:

• SQLAlchemy: the ORM layer that lets you map Python classes to database tables and write queries in Python instead of raw SQL

• PyMySQL: the database driver that handles the underlying MySQL protocol communication; SQLAlchemy does not include a driver

Install both:

pip install PyMySQL
pip install SQLAlchemy

Configure the connection

Use create_engine() to initialize a connection to your LindormTable instance.

from sqlalchemy import create_engine

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

Choosing your endpoint type:

Define a model

Define your table structure as a Python class using SQLAlchemy's declarative base.

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))

Perform CRUD operations

All examples use a SQLAlchemy session to interact with LindormTable.

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()

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

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

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

Complete sample code

The following is a complete working example. Replace the connection string placeholders with your actual values before running it.

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])

Expected output:

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

What's next

• View endpoints — get the VPC and public endpoint addresses for your instance

• Configure a whitelist — control which IP addresses can connect to your instance

• Enable the MySQL compatibility feature — enable MySQL protocol support if not already done