Skip to content

api-reference.md

Latest commit

History

History
548 lines (428 loc) · 11.2 KB

api-reference.md

File metadata and controls

548 lines (428 loc) · 11.2 KB
Raw

PySQLit API 参考文档

🎯 概述

本文档提供了 PySQLit 的完整 API 参考，包括所有公共类、方法、异常和配置选项。API 设计遵循 Python 最佳实践，提供类型提示和详细的文档字符串。

📋 目录

🏗️ 核心类

EnhancedDatabase

主数据库类，提供完整的数据库操作接口。

frompysqlit.databaseimportEnhancedDatabase# 创建数据库连接db=EnhancedDatabase("myapp.db")
# 或内存数据库db=EnhancedDatabase(":memory:")

构造函数

EnhancedDatabase(
filename: str,
page_size: int=4096,
cache_size: int=100,
timeout: float=5.0,
isolation_level: IsolationLevel=IsolationLevel.REPEATABLE_READ
) ->None

参数说明:

  • filename: 数据库文件路径，":memory:" 表示内存数据库
  • page_size: 页大小，必须是512的倍数
  • cache_size: 页缓存大小
  • timeout: 锁等待超时时间（秒）
  • isolation_level: 事务隔离级别

核心方法

基本操作
# 执行SQL语句result=db.execute(
"SELECT * FROM users WHERE age > ?",
(25,)
)
# 执行多个语句db.executemany(
"INSERT INTO users (name, email) VALUES (?, ?)",
[("Alice", "alice@example.com"), ("Bob", "bob@example.com")]
)
# 执行脚本db.executescript(""" CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT); INSERT INTO users (name) VALUES ('Charlie');""")
事务管理
# 手动事务tx_id=db.begin_transaction()
try:
db.execute("UPDATE accounts SET balance = balance - 100 WHERE id = 1")
db.execute("UPDATE accounts SET balance = balance + 100 WHERE id = 2")
db.commit_transaction(tx_id)
exceptExceptionase:
db.rollback_transaction(tx_id)
raise# 上下文管理器withdb.transaction() astx:
db.execute("INSERT INTO users VALUES (?, ?)", (1, "Alice"))
db.execute("INSERT INTO users VALUES (?, ?)", (2, "Bob"))
表管理
# 创建表db.create_table("users", {
"id": "INTEGER PRIMARY KEY AUTOINCREMENT",
"username": "TEXT UNIQUE NOT NULL",
"email": "TEXT UNIQUE NOT NULL",
"age": "INTEGER CHECK(age > 0)",
"created_at": "DATETIME DEFAULT CURRENT_TIMESTAMP"
})
# 删除表db.drop_table("users")
# 获取表信息tables=db.list_tables()
schema=db.get_table_schema("users")
备份恢复
# 创建备份backup_path=db.create_backup("daily_backup")
print(f"备份已创建: {backup_path}")
# 列出备份backups=db.list_backups()
forbackupinbackups:
print(f"{backup['name']} - {backup['created']}")
# 恢复备份db.restore_backup("daily_backup")
# 删除备份db.delete_backup("old_backup")

EnhancedTable

表操作类，提供面向对象的表操作接口。

frompysqlit.databaseimportEnhancedTable# 获取表实例table=db.get_table("users")

方法

数据操作
# 插入数据row_id=table.insert_row({
"username": "alice",
"email": "alice@example.com",
"age": 25
})
# 查询数据rows=table.select_all()
rows=table.select_with_condition(
WhereCondition("age", ">", 18)
)
# 更新数据updated=table.update_rows(
{"age": 26},
WhereCondition("username", "=", "alice")
)
# 删除数据deleted=table.delete_rows(
WhereCondition("age", "<", 18)
)
模式操作
# 添加列table.add_column("phone", "TEXT")
# 创建索引table.create_index("idx_username", ["username"], unique=True)
# 删除索引table.drop_index("idx_username")

TransactionManager

事务管理器，提供ACID事务支持。

frompysqlit.transactionimportTransactionManager, IsolationLevel# 获取事务管理器tx_manager=db.transaction_manager# 开始事务tx_id=tx_manager.begin_transaction(
isolation_level=IsolationLevel.SERIALIZABLE
)
# 提交事务tx_manager.commit_transaction(tx_id)
# 回滚事务tx_manager.rollback_transaction(tx_id)

事务隔离级别

classIsolationLevel(Enum):
READ_UNCOMMITTED="READ UNCOMMITTED"READ_COMMITTED="READ COMMITTED"REPEATABLE_READ="REPEATABLE READ"SERIALIZABLE="SERIALIZABLE"

BackupManager

备份管理器，提供高级备份功能。

frompysqlit.backupimportBackupManager# 创建备份管理器backup_mgr=BackupManager("myapp.db")
# 创建备份backup_name=backup_mgr.create_backup("manual_backup")
# 自动备份backup_thread=backup_mgr.auto_backup(interval_hours=24)
# 验证备份is_valid=backup_mgr.validate_backup("manual_backup")

🔧 数据模型

Row

行数据模型，提供字典式访问。

frompysqlit.modelsimportRow# 创建行row=Row(
id=1,
username="alice",
email="alice@example.com",
age=25
)
# 访问数据print(row["username"]) # "alice"print(row.username) # "alice"# 转换为字典data=row.to_dict()

TableSchema

表模式定义。

frompysqlit.modelsimportTableSchema, ColumnDefinition, DataType# 创建表模式schema=TableSchema("users")
schema.add_column(ColumnDefinition(
name="id",
data_type=DataType.INTEGER,
primary_key=True,
auto_increment=True
))
schema.add_column(ColumnDefinition(
name="username",
data_type=DataType.TEXT,
unique=True,
not_null=True
))

WhereCondition

WHERE条件构造器。

frompysqlit.modelsimportWhereCondition# 创建条件condition=WhereCondition("age", ">", 18)
condition=WhereCondition("username", "LIKE", "a%")
condition=WhereCondition("email", "IN", ["a@b.com", "c@d.com"])
# 复合条件frompysqlit.modelsimportAndCondition, OrConditioncomplex_condition=AndCondition([
WhereCondition("age", ">", 18),
OrCondition([
WhereCondition("status", "=", "active"),
WhereCondition("role", "=", "admin")
])
])

⚙️ 配置选项

DatabaseConfig

数据库配置类。

frompysqlit.configimportDatabaseConfigconfig=DatabaseConfig(
page_size=4096,
cache_size=100,
max_connections=10,
timeout=5.0,
isolation_level=IsolationLevel.REPEATABLE_READ,
auto_vacuum=True,
foreign_keys=True,
journal_mode="WAL"
)

配置参数

参数类型默认值说明
page_sizeint4096数据库页大小
cache_sizeint100页缓存大小
max_connectionsint10最大连接数
timeoutfloat5.0锁等待超时时间
isolation_levelIsolationLevelREPEATABLE_READ默认隔离级别
auto_vacuumboolTrue自动清理空闲页
foreign_keysboolTrue启用外键约束
journal_modestr"WAL"日志模式

🚨 异常处理

异常层次结构

PySQLitError
├── DatabaseError
│ ├── ConnectionError
│ ├── TransactionError
│ └── SchemaError
├── StorageError
│ ├── PageError
│ └── FileError
├── ParseError
│ ├── SQLSyntaxError
│ └── ValidationError
└── BackupError
├── BackupCreationError
└── BackupRestoreError

异常处理示例

frompysqlit.exceptionsimport (
DatabaseError, StorageError, ParseError, TransactionError
)
try:
db.execute("INVALID SQL")
exceptParseErrorase:
print(f"SQL语法错误: {e}")
try:
db.execute("INSERT INTO users VALUES (1, NULL)")
exceptDatabaseErrorase:
print(f"数据库错误: {e}")
try:
withdb.transaction():
db.execute("UPDATE nonexistent SET x = 1")
exceptTransactionErrorase:
print(f"事务错误: {e}")

🛠️ 工具类

ConnectionPool

连接池管理。

frompysqlit.poolimportConnectionPool# 创建连接池pool=ConnectionPool(
max_connections=10,
database_path="myapp.db",
timeout=5.0
)
# 获取连接withpool.get_connection() asdb:
users=db.execute("SELECT * FROM users LIMIT 10")
# 或使用上下文管理器withpool.context() asdb:
db.execute("INSERT INTO users VALUES (?, ?)", (1, "Alice"))

QueryBuilder

查询构建器。

frompysqlit.queryimportQueryBuilder# 构建查询query= (QueryBuilder()
.select("id", "username", "email")
.from_table("users")
.where("age", ">", 18)
.where("status", "=", "active")
.order_by("created_at", "DESC")
.limit(10)
)
# 执行查询results=db.execute(str(query), query.params)

MigrationManager

数据库迁移管理。

frompysqlit.migrationimportMigrationManager# 创建迁移migration=MigrationManager(db)
# 添加迁移migration.add_migration("001_add_users_table", """ CREATE TABLE users ( id INTEGER PRIMARY KEY AUTOINCREMENT, username TEXT UNIQUE NOT NULL, email TEXT UNIQUE NOT NULL )""")
# 执行迁移migration.migrate()

📊 性能监控

PerformanceMonitor

性能监控器。

frompysqlit.monitorimportPerformanceMonitor# 创建监控器monitor=PerformanceMonitor(db)
# 启用监控monitor.enable()
# 执行查询results=db.execute("SELECT * FROM users")
# 获取统计stats=monitor.get_stats()
print(f"查询次数: {stats['query_count']}")
print(f"平均查询时间: {stats['avg_query_time']}ms")
print(f"缓存命中率: {stats['cache_hit_rate']}%")

QueryProfiler

查询分析器。

frompysqlit.profileimportQueryProfiler# 分析查询profiler=QueryProfiler(db)
result=profiler.profile("SELECT * FROM users WHERE age > 25")
print(f"执行时间: {result.execution_time}ms")
print(f"扫描行数: {result.rows_scanned}")
print(f"使用索引: {result.index_used}")

🧪 测试工具

TestDatabase

测试专用数据库。

frompysqlit.testingimportTestDatabase# 创建测试数据库test_db=TestDatabase()
# 自动清理withtest_dbasdb:
db.execute("CREATE TABLE test (id INTEGER, name TEXT)")
db.execute("INSERT INTO test VALUES (1, 'Alice')")
# 测试结束后自动清理

MockDataGenerator

测试数据生成器。

frompysqlit.testingimportMockDataGenerator# 生成测试数据generator=MockDataGenerator(db)
generator.generate_users(count=1000)
generator.generate_posts(count=5000)

🔍 调试工具

DebugDatabase

调试模式数据库。

frompysqlit.debugimportDebugDatabase# 启用调试模式db=DebugDatabase("myapp.db", debug=True)
# 查看执行计划plan=db.explain("SELECT * FROM users WHERE age > 25")
print(plan)
# 查看锁信息locks=db.get_lock_info()
print(locks)

SQLLogger

SQL日志记录器。

frompysqlit.debugimportSQLLogger# 启用SQL日志logger=SQLLogger(db)
logger.enable()
# 执行查询db.execute("SELECT * FROM users")
# 查看日志forentryinlogger.get_logs():
print(f"{entry.timestamp}: {entry.sql} ({entry.duration}ms)")

提示: 所有API都提供了完整的类型提示和文档字符串，建议使用IDE的自动补全功能来探索更多功能！