Build a Simple CRUD App with TiDB and Python
This document describes how to use TiDB and Python to build a simple CRUD application.
Step 1. Launch your TiDB cluster
The following introduces how to start a TiDB cluster.
Use a TiDB Serverless cluster
For detailed steps, see Create a TiDB Serverless cluster.
Use a local cluster
For detailed steps, see Deploy a local test cluster or Deploy a TiDB cluster using TiUP.
Step 2. Get the code
git clone https://github.com/pingcap-inc/tidb-example-python.git
- SQLAlchemy (Recommended)
- peewee (Recommended)
- mysqlclient
- PyMySQL
- mysql-connector-python
SQLAlchemy is a popular open-source ORM library for Python. The following uses SQLAlchemy 1.44 as an example.
import uuid
from typing import List
from sqlalchemy import create_engine, String, Column, Integer, select, func
from sqlalchemy.orm import declarative_base, sessionmaker
engine = create_engine('mysql://root:@127.0.0.1:4000/test')
Base = declarative_base()
Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)
class Player(Base):
__tablename__ = "player"
id = Column(String(36), primary_key=True)
coins = Column(Integer)
goods = Column(Integer)
def __repr__(self):
return f'Player(id={self.id!r}, coins={self.coins!r}, goods={self.goods!r})'
def random_player(amount: int) -> List[Player]:
players = []
for _ in range(amount):
players.append(Player(id=uuid.uuid4(), coins=10000, goods=10000))
return players
def simple_example() -> None:
with Session() as session:
# create a player, who has a coin and a goods.
session.add(Player(id="test", coins=1, goods=1))
# get this player, and print it.
get_test_stmt = select(Player).where(Player.id == "test")
for player in session.scalars(get_test_stmt):
print(player)
# create players with bulk inserts.
# insert 1919 players totally, with 114 players per batch.
# each player has a random UUID
player_list = random_player(1919)
for idx in range(0, len(player_list), 114):
session.bulk_save_objects(player_list[idx:idx + 114])
# print the number of players
count = session.query(func.count(Player.id)).scalar()
print(f'number of players: {count}')
# print 3 players.
three_players = session.query(Player).limit(3).all()
for player in three_players:
print(player)
session.commit()
def trade_check(session: Session, sell_id: str, buy_id: str, amount: int, price: int) -> bool:
# sell player goods check
sell_player = session.query(Player.goods).filter(Player.id == sell_id).with_for_update().one()
if sell_player.goods < amount:
print(f'sell player {sell_id} goods not enough')
return False
# buy player coins check
buy_player = session.query(Player.coins).filter(Player.id == buy_id).with_for_update().one()
if buy_player.coins < price:
print(f'buy player {buy_id} coins not enough')
return False
def trade(sell_id: str, buy_id: str, amount: int, price: int) -> None:
with Session() as session:
if trade_check(session, sell_id, buy_id, amount, price) is False:
return
# deduct the goods of seller, and raise his/her the coins
session.query(Player).filter(Player.id == sell_id). \
update({'goods': Player.goods - amount, 'coins': Player.coins + price})
# deduct the coins of buyer, and raise his/her the goods
session.query(Player).filter(Player.id == buy_id). \
update({'goods': Player.goods + amount, 'coins': Player.coins - price})
session.commit()
print("trade success")
def trade_example() -> None:
with Session() as session:
# create two players
# player 1: id is "1", has only 100 coins.
# player 2: id is "2", has 114514 coins, and 20 goods.
session.add(Player(id="1", coins=100, goods=0))
session.add(Player(id="2", coins=114514, goods=20))
session.commit()
# player 1 wants to buy 10 goods from player 2.
# it will cost 500 coins, but player 1 cannot afford it.
# so this trade will fail, and nobody will lose their coins or goods
trade(sell_id="2", buy_id="1", amount=10, price=500)
# then player 1 has to reduce the incoming quantity to 2.
# this trade will be successful
trade(sell_id="2", buy_id="1", amount=2, price=100)
with Session() as session:
traders = session.query(Player).filter(Player.id.in_(("1", "2"))).all()
for player in traders:
print(player)
session.commit()
simple_example()
trade_example()
Compared with using drivers directly, SQLAlchemy provides an abstraction for the specific details of different databases when you create a database connection. In addition, SQLAlchemy encapsulates some operations such as session management and CRUD of basic objects, which greatly simplifies the code.
The Player class is a mapping of a table to attributes in the application. Each attribute of Player corresponds to a field in the player table. To provide SQLAlchemy with more information, the attribute is defined as id = Column(String(36), primary_key=True) to indicate the field type and its additional attributes. For example, id = Column(String(36), primary_key=True) indicates that the id attribute is String type, the corresponding field in database is VARCHAR type, the length is 36, and it is a primary key.
For more information about how to use SQLAlchemy, refer to SQLAlchemy documentation.
peewee is a popular open-source ORM library for Python. The following uses peewee 3.15.4 as an example.
import os
import uuid
from typing import List
from peewee import *
from playhouse.db_url import connect
db = connect('mysql://root:@127.0.0.1:4000/test')
class Player(Model):
id = CharField(max_length=36, primary_key=True)
coins = IntegerField()
goods = IntegerField()
class Meta:
database = db
table_name = "player"
def random_player(amount: int) -> List[Player]:
players = []
for _ in range(amount):
players.append(Player(id=uuid.uuid4(), coins=10000, goods=10000))
return players
def simple_example() -> None:
# create a player, who has a coin and a goods.
Player.create(id="test", coins=1, goods=1)
# get this player, and print it.
test_player = Player.select().where(Player.id == "test").get()
print(f'id:{test_player.id}, coins:{test_player.coins}, goods:{test_player.goods}')
# create players with bulk inserts.
# insert 1919 players totally, with 114 players per batch.
# each player has a random UUID
player_list = random_player(1919)
Player.bulk_create(player_list, 114)
# print the number of players
count = Player.select().count()
print(f'number of players: {count}')
# print 3 players.
three_players = Player.select().limit(3)
for player in three_players:
print(f'id:{player.id}, coins:{player.coins}, goods:{player.goods}')
def trade_check(sell_id: str, buy_id: str, amount: int, price: int) -> bool:
sell_goods = Player.select(Player.goods).where(Player.id == sell_id).get().goods
if sell_goods < amount:
print(f'sell player {sell_id} goods not enough')
return False
buy_coins = Player.select(Player.coins).where(Player.id == buy_id).get().coins
if buy_coins < price:
print(f'buy player {buy_id} coins not enough')
return False
return True
def trade(sell_id: str, buy_id: str, amount: int, price: int) -> None:
with db.atomic() as txn:
try:
if trade_check(sell_id, buy_id, amount, price) is False:
txn.rollback()
return
# deduct the goods of seller, and raise his/her the coins
Player.update(goods=Player.goods - amount, coins=Player.coins + price).where(Player.id == sell_id).execute()
# deduct the coins of buyer, and raise his/her the goods
Player.update(goods=Player.goods + amount, coins=Player.coins - price).where(Player.id == buy_id).execute()
except Exception as err:
txn.rollback()
print(f'something went wrong: {err}')
else:
txn.commit()
print("trade success")
def trade_example() -> None:
# create two players
# player 1: id is "1", has only 100 coins.
# player 2: id is "2", has 114514 coins, and 20 goods.
Player.create(id="1", coins=100, goods=0)
Player.create(id="2", coins=114514, goods=20)
# player 1 wants to buy 10 goods from player 2.
# it will cost 500 coins, but player 1 cannot afford it.
# so this trade will fail, and nobody will lose their coins or goods
trade(sell_id="2", buy_id="1", amount=10, price=500)
# then player 1 has to reduce the incoming quantity to 2.
# this trade will be successful
trade(sell_id="2", buy_id="1", amount=2, price=100)
# let's take a look for player 1 and player 2 currently
after_trade_players = Player.select().where(Player.id.in_(["1", "2"]))
for player in after_trade_players:
print(f'id:{player.id}, coins:{player.coins}, goods:{player.goods}')
db.connect()
# recreate the player table
db.drop_tables([Player])
db.create_tables([Player])
simple_example()
trade_example()
Compared with using drivers directly, peewee provides an abstraction for the specific details of different databases when you create a database connection. In addition, peewee encapsulates some operations such as session management and CRUD of basic objects, which greatly simplifies the code.
The Player class is a mapping of a table to attributes in the application. Each attribute of Player corresponds to a field in the player table. To provide SQLAlchemy with more information, the attribute is defined as id = Column(String(36), primary_key=True) to indicate the field type and its additional attributes. For example, id = Column(String(36), primary_key=True) indicates that the id attribute is String type, the corresponding field in database is VARCHAR type, the length is 36, and it is a primary key.
For more information about how to use peewee, refer to peewee documentation.
mysqlclient is a popular open-source driver for Python. The following uses mysqlclient 2.1.1 as an example. Drivers for Python are more convenient to use than other languages, but they do not shield the underlying implementation and require manual management of transactions. If there are not a lot of scenarios where SQL is required, it is recommended to use ORM, which can help reduce the coupling of your program.
import uuid
from typing import List
import MySQLdb
from MySQLdb import Connection
from MySQLdb.cursors import Cursor
def get_connection(autocommit: bool = True) -> MySQLdb.Connection:
return MySQLdb.connect(
host="127.0.0.1",
port=4000,
user="root",
password="",
database="test",
autocommit=autocommit
)
def create_player(cursor: Cursor, player: tuple) -> None:
cursor.execute("INSERT INTO player (id, coins, goods) VALUES (%s, %s, %s)", player)
def get_player(cursor: Cursor, player_id: str) -> tuple:
cursor.execute("SELECT id, coins, goods FROM player WHERE id = %s", (player_id,))
return cursor.fetchone()
def get_players_with_limit(cursor: Cursor, limit: int) -> List[tuple]:
cursor.execute("SELECT id, coins, goods FROM player LIMIT %s", (limit,))
return cursor.fetchall()
def random_player(amount: int) -> List[tuple]:
players = []
for _ in range(amount):
players.append((uuid.uuid4(), 10000, 10000))
return players
def bulk_create_player(cursor: Cursor, players: List[tuple]) -> None:
cursor.executemany("INSERT INTO player (id, coins, goods) VALUES (%s, %s, %s)", players)
def get_count(cursor: Cursor) -> None:
cursor.execute("SELECT count(*) FROM player")
return cursor.fetchone()[0]
def trade_check(cursor: Cursor, sell_id: str, buy_id: str, amount: int, price: int) -> bool:
get_player_with_lock_sql = "SELECT coins, goods FROM player WHERE id = %s FOR UPDATE"
# sell player goods check
cursor.execute(get_player_with_lock_sql, (sell_id,))
_, sell_goods = cursor.fetchone()
if sell_goods < amount:
print(f'sell player {sell_id} goods not enough')
return False
# buy player coins check
cursor.execute(get_player_with_lock_sql, (buy_id,))
buy_coins, _ = cursor.fetchone()
if buy_coins < price:
print(f'buy player {buy_id} coins not enough')
return False
def trade_update(cursor: Cursor, sell_id: str, buy_id: str, amount: int, price: int) -> None:
update_player_sql = "UPDATE player set goods = goods + %s, coins = coins + %s WHERE id = %s"
# deduct the goods of seller, and raise his/her the coins
cursor.execute(update_player_sql, (-amount, price, sell_id))
# deduct the coins of buyer, and raise his/her the goods
cursor.execute(update_player_sql, (amount, -price, buy_id))
def trade(connection: Connection, sell_id: str, buy_id: str, amount: int, price: int) -> None:
with connection.cursor() as cursor:
if trade_check(cursor, sell_id, buy_id, amount, price) is False:
connection.rollback()
return
try:
trade_update(cursor, sell_id, buy_id, amount, price)
except Exception as err:
connection.rollback()
print(f'something went wrong: {err}')
else:
connection.commit()
print("trade success")
def simple_example() -> None:
with get_connection(autocommit=True) as conn:
with conn.cursor() as cur:
# create a player, who has a coin and a goods.
create_player(cur, ("test", 1, 1))
# get this player, and print it.
test_player = get_player(cur, "test")
print(f'id:{test_player[0]}, coins:{test_player[1]}, goods:{test_player[2]}')
# create players with bulk inserts.
# insert 1919 players totally, with 114 players per batch.
# each player has a random UUID
player_list = random_player(1919)
for idx in range(0, len(player_list), 114):
bulk_create_player(cur, player_list[idx:idx + 114])
# print the number of players
count = get_count(cur)
print(f'number of players: {count}')
# print 3 players.
three_players = get_players_with_limit(cur, 3)
for player in three_players:
print(f'id:{player[0]}, coins:{player[1]}, goods:{player[2]}')
def trade_example() -> None:
with get_connection(autocommit=False) as conn:
with conn.cursor() as cur:
# create two players
# player 1: id is "1", has only 100 coins.
# player 2: id is "2", has 114514 coins, and 20 goods.
create_player(cur, ("1", 100, 0))
create_player(cur, ("2", 114514, 20))
conn.commit()
# player 1 wants to buy 10 goods from player 2.
# it will cost 500 coins, but player 1 cannot afford it.
# so this trade will fail, and nobody will lose their coins or goods
trade(conn, sell_id="2", buy_id="1", amount=10, price=500)
# then player 1 has to reduce the incoming quantity to 2.
# this trade will be successful
trade(conn, sell_id="2", buy_id="1", amount=2, price=100)
# let's take a look for player 1 and player 2 currently
with conn.cursor() as cur:
_, player1_coin, player1_goods = get_player(cur, "1")
print(f'id:1, coins:{player1_coin}, goods:{player1_goods}')
_, player2_coin, player2_goods = get_player(cur, "2")
print(f'id:2, coins:{player2_coin}, goods:{player2_goods}')
simple_example()
trade_example()
The driver has a lower level of encapsulation than ORM, so there are a lot of SQL statements in the program. Unlike ORM, there is no data object in drivers, so the Player queried by the driver is represented as a tuple.
For more information about how to use mysqlclient, refer to mysqlclient documentation.
PyMySQL is a popular open-source driver for Python. The following uses PyMySQL 1.0.2 as an example. Drivers for Python are more convenient to use than other languages, but they do not shield the underlying implementation and require manual management of transactions. If there are not a lot of scenarios where SQL is required, it is recommended to use ORM, which can help reduce the coupling of your program.
import uuid
from typing import List
import pymysql.cursors
from pymysql import Connection
from pymysql.cursors import DictCursor
def get_connection(autocommit: bool = False) -> Connection:
return pymysql.connect(host='127.0.0.1',
port=4000,
user='root',
password='',
database='test',
cursorclass=DictCursor,
autocommit=autocommit)
def create_player(cursor: DictCursor, player: tuple) -> None:
cursor.execute("INSERT INTO player (id, coins, goods) VALUES (%s, %s, %s)", player)
def get_player(cursor: DictCursor, player_id: str) -> dict:
cursor.execute("SELECT id, coins, goods FROM player WHERE id = %s", (player_id,))
return cursor.fetchone()
def get_players_with_limit(cursor: DictCursor, limit: int) -> tuple:
cursor.execute("SELECT id, coins, goods FROM player LIMIT %s", (limit,))
return cursor.fetchall()
def random_player(amount: int) -> List[tuple]:
players = []
for _ in range(amount):
players.append((uuid.uuid4(), 10000, 10000))
return players
def bulk_create_player(cursor: DictCursor, players: List[tuple]) -> None:
cursor.executemany("INSERT INTO player (id, coins, goods) VALUES (%s, %s, %s)", players)
def get_count(cursor: DictCursor) -> int:
cursor.execute("SELECT count(*) as count FROM player")
return cursor.fetchone()['count']
def trade_check(cursor: DictCursor, sell_id: str, buy_id: str, amount: int, price: int) -> bool:
get_player_with_lock_sql = "SELECT coins, goods FROM player WHERE id = %s FOR UPDATE"
# sell player goods check
cursor.execute(get_player_with_lock_sql, (sell_id,))
seller = cursor.fetchone()
if seller['goods'] < amount:
print(f'sell player {sell_id} goods not enough')
return False
# buy player coins check
cursor.execute(get_player_with_lock_sql, (buy_id,))
buyer = cursor.fetchone()
if buyer['coins'] < price:
print(f'buy player {buy_id} coins not enough')
return False
def trade_update(cursor: DictCursor, sell_id: str, buy_id: str, amount: int, price: int) -> None:
update_player_sql = "UPDATE player set goods = goods + %s, coins = coins + %s WHERE id = %s"
# deduct the goods of seller, and raise his/her the coins
cursor.execute(update_player_sql, (-amount, price, sell_id))
# deduct the coins of buyer, and raise his/her the goods
cursor.execute(update_player_sql, (amount, -price, buy_id))
def trade(connection: Connection, sell_id: str, buy_id: str, amount: int, price: int) -> None:
with connection.cursor() as cursor:
if trade_check(cursor, sell_id, buy_id, amount, price) is False:
connection.rollback()
return
try:
trade_update(cursor, sell_id, buy_id, amount, price)
except Exception as err:
connection.rollback()
print(f'something went wrong: {err}')
else:
connection.commit()
print("trade success")
def simple_example() -> None:
with get_connection(autocommit=True) as connection:
with connection.cursor() as cur:
# create a player, who has a coin and a goods.
create_player(cur, ("test", 1, 1))
# get this player, and print it.
test_player = get_player(cur, "test")
print(test_player)
# create players with bulk inserts.
# insert 1919 players totally, with 114 players per batch.
# each player has a random UUID
player_list = random_player(1919)
for idx in range(0, len(player_list), 114):
bulk_create_player(cur, player_list[idx:idx + 114])
# print the number of players
count = get_count(cur)
print(f'number of players: {count}')
# print 3 players.
three_players = get_players_with_limit(cur, 3)
for player in three_players:
print(player)
def trade_example() -> None:
with get_connection(autocommit=False) as connection:
with connection.cursor() as cur:
# create two players
# player 1: id is "1", has only 100 coins.
# player 2: id is "2", has 114514 coins, and 20 goods.
create_player(cur, ("1", 100, 0))
create_player(cur, ("2", 114514, 20))
connection.commit()
# player 1 wants to buy 10 goods from player 2.
# it will cost 500 coins, but player 1 cannot afford it.
# so this trade will fail, and nobody will lose their coins or goods
trade(connection, sell_id="2", buy_id="1", amount=10, price=500)
# then player 1 has to reduce the incoming quantity to 2.
# this trade will be successful
trade(connection, sell_id="2", buy_id="1", amount=2, price=100)
# let's take a look for player 1 and player 2 currently
with connection.cursor() as cur:
print(get_player(cur, "1"))
print(get_player(cur, "2"))
simple_example()
trade_example()
The driver has a lower level of encapsulation than ORM, so there are a lot of SQL statements in the program. Unlike ORM, there is no data object in drivers, so the Player queried by the driver is represented as a dictionary.
For more information about how to use PyMySQL, refer to PyMySQL documentation.
mysql-connector-python is a popular open-source driver for Python. The following uses mysql-connector-python 8.0.31 as an example. Drivers for Python are more convenient to use than other languages, but they do not shield the underlying implementation and require manual management of transactions. If there are not a lot of scenarios where SQL is required, it is recommended to use ORM, which can help reduce the coupling of your program.
import uuid
from typing import List
from mysql.connector import connect, MySQLConnection
from mysql.connector.cursor import MySQLCursor
def get_connection(autocommit: bool = True) -> MySQLConnection:
connection = connect(host='127.0.0.1',
port=4000,
user='root',
password='',
database='test')
connection.autocommit = autocommit
return connection
def create_player(cursor: MySQLCursor, player: tuple) -> None:
cursor.execute("INSERT INTO player (id, coins, goods) VALUES (%s, %s, %s)", player)
def get_player(cursor: MySQLCursor, player_id: str) -> tuple:
cursor.execute("SELECT id, coins, goods FROM player WHERE id = %s", (player_id,))
return cursor.fetchone()
def get_players_with_limit(cursor: MySQLCursor, limit: int) -> List[tuple]:
cursor.execute("SELECT id, coins, goods FROM player LIMIT %s", (limit,))
return cursor.fetchall()
def random_player(amount: int) -> List[tuple]:
players = []
for _ in range(amount):
players.append((str(uuid.uuid4()), 10000, 10000))
return players
def bulk_create_player(cursor: MySQLCursor, players: List[tuple]) -> None:
cursor.executemany("INSERT INTO player (id, coins, goods) VALUES (%s, %s, %s)", players)
def get_count(cursor: MySQLCursor) -> int:
cursor.execute("SELECT count(*) FROM player")
return cursor.fetchone()[0]
def trade_check(cursor: MySQLCursor, sell_id: str, buy_id: str, amount: int, price: int) -> bool:
get_player_with_lock_sql = "SELECT coins, goods FROM player WHERE id = %s FOR UPDATE"
# sell player goods check
cursor.execute(get_player_with_lock_sql, (sell_id,))
_, sell_goods = cursor.fetchone()
if sell_goods < amount:
print(f'sell player {sell_id} goods not enough')
return False
# buy player coins check
cursor.execute(get_player_with_lock_sql, (buy_id,))
buy_coins, _ = cursor.fetchone()
if buy_coins < price:
print(f'buy player {buy_id} coins not enough')
return False
def trade_update(cursor: MySQLCursor, sell_id: str, buy_id: str, amount: int, price: int) -> None:
update_player_sql = "UPDATE player set goods = goods + %s, coins = coins + %s WHERE id = %s"
# deduct the goods of seller, and raise his/her the coins
cursor.execute(update_player_sql, (-amount, price, sell_id))
# deduct the coins of buyer, and raise his/her the goods
cursor.execute(update_player_sql, (amount, -price, buy_id))
def trade(connection: MySQLConnection, sell_id: str, buy_id: str, amount: int, price: int) -> None:
with connection.cursor() as cursor:
if trade_check(cursor, sell_id, buy_id, amount, price) is False:
connection.rollback()
return
try:
trade_update(cursor, sell_id, buy_id, amount, price)
except Exception as err:
connection.rollback()
print(f'something went wrong: {err}')
else:
connection.commit()
print("trade success")
def simple_example() -> None:
with get_connection(autocommit=True) as connection:
with connection.cursor() as cur:
# create a player, who has a coin and a goods.
create_player(cur, ("test", 1, 1))
# get this player, and print it.
test_player = get_player(cur, "test")
print(f'id:{test_player[0]}, coins:{test_player[1]}, goods:{test_player[2]}')
# create players with bulk inserts.
# insert 1919 players totally, with 114 players per batch.
# each player has a random UUID
player_list = random_player(1919)
for idx in range(0, len(player_list), 114):
bulk_create_player(cur, player_list[idx:idx + 114])
# print the number of players
count = get_count(cur)
print(f'number of players: {count}')
# print 3 players.
three_players = get_players_with_limit(cur, 3)
for player in three_players:
print(f'id:{player[0]}, coins:{player[1]}, goods:{player[2]}')
def trade_example() -> None:
with get_connection(autocommit=False) as conn:
with conn.cursor() as cur:
# create two players
# player 1: id is "1", has only 100 coins.
# player 2: id is "2", has 114514 coins, and 20 goods.
create_player(cur, ("1", 100, 0))
create_player(cur, ("2", 114514, 20))
conn.commit()
# player 1 wants to buy 10 goods from player 2.
# it will cost 500 coins, but player 1 cannot afford it.
# so this trade will fail, and nobody will lose their coins or goods
trade(conn, sell_id="2", buy_id="1", amount=10, price=500)
# then player 1 has to reduce the incoming quantity to 2.
# this trade will be successful
trade(conn, sell_id="2", buy_id="1", amount=2, price=100)
# let's take a look for player 1 and player 2 currently
with conn.cursor() as cur:
_, player1_coin, player1_goods = get_player(cur, "1")
print(f'id:1, coins:{player1_coin}, goods:{player1_goods}')
_, player2_coin, player2_goods = get_player(cur, "2")
print(f'id:2, coins:{player2_coin}, goods:{player2_goods}')
simple_example()
trade_example()
The driver has a lower level of encapsulation than ORM, so there are a lot of SQL statements in the program. Unlike ORM, there is no data object in drivers, so the Player queried by the driver is represented as a tuple.
For more information about how to use mysql-connector-python, refer to mysql-connector-python documentation.
Step 3. Run the code
The following content introduces how to run the code step by step.
Step 3.1 Initialize table
Before running the code, you need to initialize the table manually. If you are using a local TiDB cluster, you can run the following command:
- MySQL CLI
- MyCLI
mysql --host 127.0.0.1 --port 4000 -u root < player_init.sql
mycli --host 127.0.0.1 --port 4000 -u root --no-warn < player_init.sql
If you are not using a local cluster, or have not installed a MySQL client, connect to your cluster using your preferred method (such as Navicat, DBeaver, or other GUI tools) and run the SQL statements in the player_init.sql file.
Step 3.2 Modify parameters for TiDB Cloud
If you are using a TiDB Serverless cluster, you need to provide your CA root path and replace <ca_path> in the following examples with your CA path. To get the CA root path on your system, refer to Where is the CA root path on my system?.
- SQLAlchemy (Recommended)
- peewee (Recommended)
- mysqlclient
- PyMySQL
- mysql-connector-python
If you are using a TiDB Serverless cluster, modify the parameters of the create_engine function in sqlalchemy_example.py:
engine = create_engine('mysql://root:@127.0.0.1:4000/test')
Suppose that the password you set is 123456, and the connection parameters you get from the cluster details page are the following:
- Endpoint:
xxx.tidbcloud.com - Port:
4000 - User:
2aEp24QWEDLqRFs.root
In this case, you can modify the create_engine as follows:
engine = create_engine('mysql://2aEp24QWEDLqRFs.root:123456@xxx.tidbcloud.com:4000/test', connect_args={
"ssl_mode": "VERIFY_IDENTITY",
"ssl": {
"ca": "<ca_path>"
}
})
If you are using a TiDB Serverless cluster, modify the parameters of the create_engine function in sqlalchemy_example.py:
db = connect('mysql://root:@127.0.0.1:4000/test')
Suppose that the password you set is 123456, and the connection parameters you get from the cluster details page are the following:
- Endpoint:
xxx.tidbcloud.com - Port:
4000 - User:
2aEp24QWEDLqRFs.root
In this case, you can modify the connect as follows:
When peewee uses PyMySQL as the driver:
db = connect('mysql://2aEp24QWEDLqRFs.root:123456@xxx.tidbcloud.com:4000/test', ssl_verify_cert=True, ssl_ca="<ca_path>")When peewee uses mysqlclient as the driver:
db = connect('mysql://2aEp24QWEDLqRFs.root:123456@xxx.tidbcloud.com:4000/test', ssl_mode="VERIFY_IDENTITY", ssl={"ca": "<ca_path>"})
Because peewee will pass parameters to the driver, you need to pay attention to the usage type of the driver when using peewee.
If you are using a TiDB Serverless cluster, change the get_connection function in mysqlclient_example.py:
def get_connection(autocommit: bool = True) -> MySQLdb.Connection:
return MySQLdb.connect(
host="127.0.0.1",
port=4000,
user="root",
password="",
database="test",
autocommit=autocommit
)
Suppose that the password you set is 123456, and the connection parameters you get from the cluster details page are the following:
- Endpoint:
xxx.tidbcloud.com - Port:
4000 - User:
2aEp24QWEDLqRFs.root
In this case, you can modify the get_connection as follows:
def get_connection(autocommit: bool = True) -> MySQLdb.Connection:
return MySQLdb.connect(
host="xxx.tidbcloud.com",
port=4000,
user="2aEp24QWEDLqRFs.root",
password="123456",
database="test",
autocommit=autocommit,
ssl_mode="VERIFY_IDENTITY",
ssl={
"ca": "<ca_path>"
}
)
If you are using a TiDB Serverless cluster, change the get_connection function in pymysql_example.py:
def get_connection(autocommit: bool = False) -> Connection:
return pymysql.connect(host='127.0.0.1',
port=4000,
user='root',
password='',
database='test',
cursorclass=DictCursor,
autocommit=autocommit)
Suppose that the password you set is 123456, and the connection parameters you get from the cluster details page are the following:
- Endpoint:
xxx.tidbcloud.com - Port:
4000 - User:
2aEp24QWEDLqRFs.root
In this case, you can modify the get_connection as follows:
def get_connection(autocommit: bool = False) -> Connection:
return pymysql.connect(host='xxx.tidbcloud.com',
port=4000,
user='2aEp24QWEDLqRFs.root',
password='123546',
database='test',
cursorclass=DictCursor,
autocommit=autocommit,
ssl_ca='<ca_path>',
ssl_verify_cert=True,
ssl_verify_identity=True)
If you are using a TiDB Serverless cluster, change the get_connection function in mysql_connector_python_example.py:
def get_connection(autocommit: bool = True) -> MySQLConnection:
connection = connect(host='127.0.0.1',
port=4000,
user='root',
password='',
database='test')
connection.autocommit = autocommit
return connection
Suppose that the password you set is 123456, and the connection parameters you get from the cluster details page are the following:
- Endpoint:
xxx.tidbcloud.com - Port:
4000 - User:
2aEp24QWEDLqRFs.root
In this case, you can modify the get_connection as follows:
def get_connection(autocommit: bool = True) -> MySQLConnection:
connection = connect(
host="xxx.tidbcloud.com",
port=4000,
user="2aEp24QWEDLqRFs.root",
password="123456",
database="test",
autocommit=autocommit,
ssl_ca='<ca_path>',
ssl_verify_identity=True
)
connection.autocommit = autocommit
return connection
Step 3.3 Run the code
Before running the code, use the following command to install dependencies:
pip3 install -r requirement.txt
If you need to run the script multiple times, follow the Table initialization section to initialize the table again before each run.
- SQLAlchemy (Recommended)
- peewee (Recommended)
- mysqlclient
- PyMySQL
- mysql-connector-python
python3 sqlalchemy_example.py
python3 peewee_example.py
python3 mysqlclient_example.py
python3 pymysql_example.py
python3 mysql_connector_python_example.py
Step 4. Expected output
- SQLAlchemy (Recommended)
- peewee (Recommended)
- mysqlclient
- PyMySQL
- mysql-connector-python