第五章：Python 集成¶

基本使用¶

连接数据库¶

import sqlite3

# 连接数据库（不存在则创建）
conn = sqlite3.connect('example.db')

# 创建游标
cursor = conn.cursor()

# 关闭连接
conn.close()

创建表¶

cursor.execute('''
    CREATE TABLE IF NOT EXISTS users (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        name TEXT NOT NULL,
        email TEXT UNIQUE
    )
''')
conn.commit()

CRUD 操作¶

# 插入
cursor.execute(
    "INSERT INTO users (name, email) VALUES (?, ?)",
    ('Alice', 'alice@example.com')
)
conn.commit()

# 查询
cursor.execute("SELECT * FROM users")
rows = cursor.fetchall()
for row in rows:
    print(row)

# 更新
cursor.execute(
    "UPDATE users SET name = ? WHERE id = ?",
    ('Alice2', 1)
)
conn.commit()

# 删除
cursor.execute("DELETE FROM users WHERE id = ?", (1,))
conn.commit()

使用 ORM¶

SQLAlchemy¶

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

engine = create_engine('sqlite:///example.db')
Base = declarative_base()

class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String)
    email = Column(String)

Base.metadata.create_all(engine)

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

# 添加用户
user = User(name='Alice', email='alice@example.com')
session.add(user)
session.commit()

异步操作¶

aiosqlite¶

import aiosqlite

async def main():
    async with aiosqlite.connect('example.db') as db:
        async with db.execute("SELECT * FROM users") as cursor:
            async for row in cursor:
                print(row)

# 运行
import asyncio
asyncio.run(main())

小结¶

Python 集成要点：

基本使用：sqlite3 模块
CRUD 操作：execute()、fetchall()
ORM：SQLAlchemy
异步：aiosqlite

下一章我们将学习事务处理。