Connect to TiDB Cloud Lake Using Python

Connect to TiDB Cloud Lake using Python with our official drivers supporting both synchronous and asynchronous operations.

Quick Start

Choose your preferred approach:

Connection String: See driver overview for DSN format and examples.

tidbcloudlake-driver

Features

• ✅ Native Performance: Direct connection to TiDB Cloud Lake
• ✅ Async/Sync Support: Choose your programming style
• ✅ PEP 249 Compatible: Standard Python DB API
• ✅ Type Safety: Full Python type mapping

Synchronous Usage

from tidbcloudlake_driver import BlockingLakeClient

# Connect and execute
client = BlockingLakeClient('<your-dsn>')
cursor = client.cursor()

# DDL: Create table
cursor.execute("CREATE TABLE users (id INT, name STRING)")

# Write: Insert data
cursor.execute("INSERT INTO users VALUES (?, ?)", (1, 'Alice'))

# Query: Read data
# Query: Read data
cursor.execute("SELECT * FROM users")

# Get column names
# cursor.description returns a list of tuples, where the first element is the column name
print(f"Columns: {[desc[0] for desc in cursor.description]}")

for row in cursor.fetchall():
    # row is a tidbcloudlake_driver.Row object
    # Access by column name
    print(f"id: {row['id']}, name: {row['name']}")

cursor.close()

Working with Row Objects

The Row object supports multiple access patterns and methods:

for row in cursor.fetchall():
    # 1. Access by column name (Recommended)
    print(f"Name: {row['name']}")

    # 2. Access by index
    print(f"First column: {row[0]}")

    # 3. Convert to tuple
    print(f"Values: {row.values()}")

    # 4. Explicit methods
    print(row.get_by_field('name'))
    print(row.get_by_index(0))

Asynchronous Usage

import asyncio
from tidbcloudlake_driver import AsyncLakeClient

async def main():
    client = AsyncLakeClient('lake://root:root@localhost:8000/?sslmode=disable')
    conn = await client.get_conn()

    # DDL: Create table
    await conn.exec("CREATE TABLE users (id INT, name STRING)")

    # Write: Insert data
    await conn.exec("INSERT INTO users VALUES (?, ?)", (1, 'Alice'))

    # Query: Read data
    rows = await conn.query_iter("SELECT * FROM users")
    async for row in rows:
        print(row.values())

    await conn.close()

asyncio.run(main())

Data Type Mappings

Resources

• PyPI: tidbcloudlake-driver
• GitHub: tidbcloudlake-driver