Turso Python 绑定pyturso实战指南DB-API 2.0 驱动、asyncio 与远程同步【免费下载链接】tursoA SQL database in Rust: SQLite-compatible, now also speaking Postgres (experimental). The LLVM of databases.项目地址: https://gitcode.com/GitHub_Trending/tu/tursopyturso 是 Turso 数据库用 Rust 编写的 SQLite 兼容嵌入式数据库引擎的官方 Python 绑定将引擎直接嵌入 Python 进程无需网络开销即可使用 SQLite 查询语言与文件格式并在此基础上提供可选的远程同步能力与 asyncio 非阻塞访问。读完本文你将掌握 pyturso 的安装方式、标准 DB-API 2.0 驱动的完整用法、基于工作线程的 asyncio 异步 API以及 pull/push 双向往远程 Turso 数据库同步、部分引导partial bootstrap等企业级离线优先方案。项目概览与核心特性Turso 是一个运行在进程内的 SQL 数据库由 Rust 编写、与 SQLite 兼容。pyturso 作为其 Python 绑定目前处于 Beta 阶段pyproject.toml中声明Development Status :: 4 - Beta在生产环境中已被多家组织使用但尚未达到 1.0 版本因此官方建议与任何数据库一样保留备份。关于 SQLite 查询语言和文件格式的兼容性状态可参考仓库根目录的 COMPAT.md。根据 bindings/python/README.mdpyturso 提供以下核心能力SQLite 兼容支持 SQLite 查询语言与文件格式进程内运行无网络开销直接在你的 Python 进程中运行跨平台支持 Linux、macOS、Windows见 pyproject.toml 中的操作系统分类器远程部分同步可从远程数据库引导本地状态、拉取远端变更、在联网时推送本地变更——同时保证离线状态下数据库完全可用asyncio 支持内置 asyncio 集成确保查询不会阻塞事件循环。安装与环境要求pyturso 通过 PyPI 以pyturso包名分发官方推荐使用 uv 安装uv pip install pyturso根据 pyproject.toml安装与运行需要满足以下前提Python 版本requires-python 3.10分类器覆盖 Python 3.10 至 3.14运行时依赖仅typing-extensions 4.6.0,!4.7.0依赖面非常小构建后端maturinRust/Python 绑定构建工具模块名称为turso._turso使用 PyO3 绑定可选依赖pyturso[sqlalchemy]会安装sqlalchemy2.0.45用于启用 SQLAlchemy 方言支持。开发与测试依赖dev组包括 mypy、pytest、pytest-cov、ruff、coverage 与 maturin便于在仓库内运行pytest测试路径配置为tests目录见[tool.pytest.ini_options]。数据库驱动标准 DB-API 2.0 用法pyturso 的核心驱动实现了 PEP 249DB-API 2.0规范接口风格与标准库sqlite3高度相似上手成本极低。以下是最小可用示例内存数据库import turso # Standard DB-API usage conn turso.connect(:memory:) cur conn.cursor() cur.execute(CREATE TABLE users (id INTEGER PRIMARY KEY, username TEXT)) cur.execute(INSERT INTO users VALUES (1, alice), (2, bob)) cur.execute(SELECT * FROM users ORDER BY id) rows cur.fetchall() print(rows) # [(1, alice), (2, bob)] conn.close()turso.connect()的第一个参数database可以是内存数据库:memory:也可以是磁盘文件路径。除此之外lib.py 中的connect()还支持以下关键字参数参数说明experimental_features逗号分隔的待启用特性列表vfs指定虚拟文件系统VFS实现encryptionEncryptionOpts对象cipherhexkey用于打开加密数据库isolation_level事务隔离级别DEFERRED默认、IMMEDIATE、EXCLUSIVE或None禁用隐式事务extra_io内部使用的 IO 回调钩子同步驱动会用它驱动底层引擎DB-API 2.0 模块属性在 lib.py 中声明了三个标准模块级属性apilevel 2.0实现 DB-API 2.0 规范threadsafety 1线程可以共享模块但不能共享连接对象paramstyle qmark仅支持?位置参数占位符。此外模块还导出了sqlite_version与sqlite_version_info通过临时连接执行SELECT sqlite_version()获取失败时回退到 Turso 跟踪的 SQLite 版本以及完整的异常类层次Warning、Error、InterfaceError、DatabaseError、DataError、OperationalError、IntegrityError、InternalError、ProgrammingError、NotSupportedError。这些异常由 lib.py 中的_map_turso_exception()将引擎底层错误映射为 DB-API 规范异常例如Busy/Interrupt→OperationalErrorConstraint→IntegrityError。Connection事务、中断与超时Connection对象lib.py模仿sqlite3.Connection的常用 API并补充了几个 Turso 扩展方法commit()/rollback()遵循 PEP 249 语义在autocommitFalse模式下提交或回滚后会自动重新开启事务_ensure_transaction_openautocommit属性可设为TrueSQLite 自动提交模式commit/rollback 为 no-op、FalsePEP 249 模式始终保证事务开启或LEGACYsqlite3 旧式隐式事务对INSERT/UPDATE/DELETE/REPLACE等 DML 自动BEGINin_transaction当前是否处于事务中interrupt()从另一个线程中止当前执行中的查询底层执行会释放 GIL因此看门狗线程可以真正运行被中断的调用会抛出OperationalError(interrupted)set_query_timeout(milliseconds)/get_query_timeout()Turso 扩展设置单条语句的最长执行时间毫秒超时后由引擎内部强制中断并抛出OperationalError0表示禁用。与interrupt()不同该功能不需要看门狗线程上下文管理器with turso.connect(...) as conn:在无异常时自动 commit有异常时自动 rollback并始终向上传播异常batch()批量执行多条参数化语句详见下文。Cursor执行与参数绑定Cursorlib.py支持标准 DB-API 方法execute、executemany、executescript、fetchone、fetchmany(size)默认取arraysize初始为 1、fetchall、setinputsizes/setoutputsizeDB-API 兼容性 no-op并可迭代__iter__/__next__逐行返回。参数绑定规则与sqlite3行为对齐位置参数使用?占位符传入元组或列表命名参数支持:name、name、$name三种前缀传入Mapping字典映射中多余的键会被忽略缺失的键由引擎报错数字参数?NNN可用字符串键1绑定到?1混合规则?位置占位符不能与映射混用。executemany()限制与sqlite3一致仅接受单条 DML 语句INSERT/UPDATE/DELETE/REPLACE否则抛出ProgrammingErrorrowcount反映最后一条语句的修改行数lastrowid保持不变。Row对象turso.Row是类似sqlite3.Row的容器支持按索引和列名访问row[0]、row[username]、keys()、切片、哈希与比较运算。注意它不能与标准库sqlite3.Row混用——若将sqlite3.Row作为row_factory传入会抛出TypeError并提示改用turso.Row。executescript()按语句拆分脚本逐一执行并丢弃结果行在 LEGACY 模式下若已有挂起事务会先隐式 COMMIT与sqlite3行为一致。batch事务化批量执行Connection.batch()lib.py是 Turso 特有的高效批量接口results conn.batch([ INSERT INTO users VALUES (1, alice), (INSERT INTO users VALUES (?, ?), (2, bob)), ], modedeferred)每条语句可以是 SQL 字符串也可以是(sql, parameters)二元组语句按顺序执行遇到第一条失败语句即停止异常携带batch_index失败的零基索引与batch_results每条语句的结果或Nonemode取deferred、immediate、exclusive、concurrent时整个批次包裹在BEGIN mode/COMMIT中失败则ROLLBACK实现全有或全无的事务语义此时批次内不允许出现事务控制语句BEGIN/COMMIT/ROLLBACK/SAVEPOINT等mode为None时批次不开启事务每条语句独立提交返回结果为一个BatchResult列表包含rows、description、rowcount、lastrowidrows_read/rows_written/query_duration_ms为服务端统计字段嵌入式引擎下为None若连接上已有开启的事务批次语句会加入该事务且mode被忽略。加密数据库与日志turso.EncryptionOpts(cipher, hexkey)用于打开加密数据库仓库中的 examples/python/encryption.py 演示了完整用法。turso.setup_logging(level)可将引擎日志接入 Python 标准logging模块将 Rust 侧TRACE/DEBUG/INFO/WARN/ERROR映射为 Python 日志级别便于调试import logging import turso turso.setup_logging(logging.DEBUG)asyncio 驱动非阻塞数据库访问turso.aio提供完全异步的 API查询在后台工作线程执行不会阻塞事件循环import asyncio import turso.aio async def main(): # Connect and use as an async context manager async with turso.aio.connect(:memory:) as conn: # Executes multiple statements await conn.executescript( CREATE TABLE t (id INTEGER PRIMARY KEY, name TEXT); INSERT INTO t(name) VALUES (alice), (bob); ) # Use a cursor for parameterized queries cur conn.cursor() await cur.execute(SELECT COUNT(*) FROM t WHERE name LIKE ?, (a%,)) count (await cur.fetchone())[0] print(count) # 1 # JSON and generate_series also available cur conn.cursor() await cur.execute(SELECT SUM(value) FROM generate_series(1, 10)) print((await cur.fetchone())[0]) # 55 asyncio.run(main())turso.aio.connect()与阻塞版签名一致同样支持experimental_features、isolation_level、extra_io但它返回一个可等待的Connection——在async with或await时才真正建立底层连接。实现原理异步 API 基于工作线程 事件循环模型。由 worker.py 中的Worker线程消费无界SimpleQueue中的(future, callable)任务顺序执行数据库操作再通过loop.call_soon_threadsafe将结果或异常交回事件循环线程。这样即使引擎执行时释放 GIL阻塞操作也不会卡住事件循环。异步Connection/Cursorlib_aio.py以缓存属性isolation_level、row_factory、text_factory、autocommit镜像底层连接状态并同样支持batch()。异步游标 API 与阻塞版一一对应await cur.execute(...)、await cur.fetchone()、fetchmany、fetchall、executemany、executescript且可作为异步上下文管理器使用async with cur:结束时自动关闭。同步驱动本地 远程的离线优先方案turso.sync允许你在本地使用远程 Turso 数据库从远端引导本地状态、拉取远端变更、推送本地提交。使用前提是你已拥有一个 Turso 远程 URL数据库的 URL 形如https://db.region.turso.io并完成相应的开通与鉴权。import turso.sync # Connect a local database to a remote Turso database conn turso.sync.connect( :memory:, # local db path (or a file path) remote_urlhttps://db.region.turso.io # your remote URL ) # Read data (fetched from remote if not present locally yet) rows conn.execute(SELECT * FROM t).fetchall() print(rows) # Pull new changes from remote into local changed conn.pull() print(Pulled:, changed) # True if there were new remote changes # Make local changes conn.execute(INSERT INTO t VALUES (push works)) conn.commit() # Push local commits to remote conn.push() # Optional: inspect and manage sync state stats conn.stats() print(Network received (bytes):, stats.network_received_bytes) conn.checkpoint() # compact local WAL after many writes conn.close()turso.sync.ConnectionSynclib_sync.py继承自嵌入式Connection在其基础上新增四个同步方法pull() - bool等待并应用远端变更若有新变更返回True否则返回Falsepush()将本地已提交的变更推送到远端checkpoint()对同步数据库的 WAL 执行检查点压缩本地 WAL大量写入后建议执行stats()返回同步数据库统计信息如network_received_bytes网络接收字节数、main_wal_size主 WAL 大小等。由于继承自标准Connection同步连接同样支持execute、commit、batch等全部 DB-API 能力且本地数据库路径既可以是:memory:也可以是磁盘文件路径。connect_sync 完整参数turso.sync.connect即connect_sync见 lib_sync.py的参数比普通connect丰富得多参数说明path本地主数据库文件路径或:memory:remote_url远程 Turso 数据库 URL可以是字符串或每次请求时求值的 callablelambda: get_current_url()支持libsql://、turso://前缀并自动转换为https://auth_token可选鉴权令牌同样支持字符串或 callable动态刷新令牌场景非常有用以Authorization: Bearer token头发送client_name客户端标识默认turso-sync-py会写入 HTTP User-Agent 便于服务端日志排查long_poll_timeout_mspull 时长轮询的超时时间毫秒bootstrap_if_empty若本地为空库create()时自动从远端引导数据默认Truepartial_sync_experimental实验性部分同步配置见下节experimental_features、isolation_level透传给底层连接remote_encryption_key加密 Turso Cloud 数据库的 base64 编码加密密钥remote_encryption_cipher远端加密算法用于计算 reserved_bytespush_operations_threshold单个 push HTTP 批次中 CDC 操作数的上限按事务边界拆分单个事务不会被拆开None时一次发送全部变更集pull_bytes_threshold引导下载拆分提示将引导拆分为多个不小于该字节数的 pull-updates HTTP 请求None时单次往返完成引导查询引导策略下无效logical_mvcc_pull同步协议覆盖开关None默认根据首次响应自动检测并持久化远端协议True强制使用 MVCC 逻辑日志流False强制页面流仅供测试或逃生通道使用网络传输实现细节同步引擎的 IO 由 lib_sync.py 中的 HTTP 处理器完成——以 64 KiB 分块流式读写响应、自动附加Authorization: Bearer与User-Agent头本地文件全量读写在写入时采用临时文件 os.replace原子重命名策略避免并发读者观察到半写状态任何 IO 失败都会 poison 对应 IO 项并最终以映射后的 DB-API 异常抛出。部分引导Partial Bootstrap默认情况下首次连接会拉取整个数据库。对于大型数据库可以使用实验性的部分同步功能仅先获取少量前缀数据后续按需补充从而显著降低初始网络成本import turso.sync conn turso.sync.connect( local.db, remote_urlhttps://db.region.turso.io, # fetch first 128 KiB upfront partial_sync_experimentalturso.sync.PartialSyncOpts( bootstrap_strategyturso.sync.PartialSyncPrefixBootstrap(length128 * 1024), ), )PartialSyncOpts支持两种引导策略lib_sync.pyPartialSyncPrefixBootstrap(length)先获取数据库文件前 N 字节N 为int适合只需要数据库开头部分数据即可启动的场景PartialSyncQueryBootstrap(query)按给定 SQL 查询在服务端命中fetch的页面进行引导只下载该查询实际访问的数据页。此外PartialSyncOpts还提供segment_size分段大小与prefetch是否预取两个可选调优参数。同步驱动asyncio 版本turso.aio.sync将上述同步原语包装为完全异步的 APIimport asyncio async def main(): conn await turso.aio.sync.connect(:memory:, remote_urlhttps://db.region.turso.io) # Read data rows await (await conn.execute(SELECT * FROM t)).fetchall() print(rows) # Pull and push await conn.pull() await conn.execute(INSERT INTO t VALUES (hello from asyncio)) await conn.commit() await conn.push() # Stats and maintenance stats await conn.stats() print(Main WAL size:, stats.main_wal_size) await conn.checkpoint() await conn.close() asyncio.run(main())其实现lib_sync_aio.py复用异步Connection的工作线程模型阻塞的同步连接在 worker 线程中创建与持有pull/push/checkpoint/stats通过_run()调度到该线程执行并await结果。connect返回可等待对象同样支持async with上下文管理器。remote_url、auth_token、partial_sync_experimental等参数与阻塞版完全一致。SQLAlchemy 集成pyturso 提供三个 SQLAlchemy 方言详见 SQLALCHEMY_DIALECT.md方言注册见 pyproject.toml 中的 entry pointssqliteturso://本地连接同步引擎sqliteaioturso://本地连接SQLAlchemy 异步引擎sqliteturso_sync://支持远程同步的连接。安装pip install pyturso[sqlalchemy]后即可使用要求 SQLAlchemy ≥ 2.0.45from sqlalchemy import create_engine, text # In-memory database engine create_engine(sqliteturso:///:memory:) # File-based database engine create_engine(sqliteturso:///path/to/database.db) with engine.connect() as conn: conn.execute(text(CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT))) conn.execute(text(INSERT INTO users (name) VALUES (Alice))) conn.commit() result conn.execute(text(SELECT * FROM users)) for row in result: print(row)同步方言通过 URL 查询参数或connect_args传入远程配置auth_token支持 callable 以便动态刷新令牌from sqlalchemy import create_engine, text from turso.sqlalchemy import get_sync_connection # Via URL query parameters engine create_engine( sqliteturso_sync:///local.db ?remote_urlhttps://your-db.turso.io auth_tokenyour-token ) with engine.connect() as conn: sync get_sync_connection(conn) # exposes the underlying sync engine sync.pull() # Pull changes from remote result conn.execute(text(SELECT * FROM users)) conn.execute(text(INSERT INTO users (name) VALUES (Bob))) conn.commit() sync.push() # Push changes to remoteURL 中支持的查询参数包括isolation_levelDEFERRED/IMMEDIATE/EXCLUSIVE/AUTOCOMMIT、experimental_features逗号分隔特性标志、remote_url同步方言必填与auth_token。异步方言示例与 ORMdeclarative Session用法同样收录在 SQLALCHEMY_DIALECT.md 中。测试与行为验证仓库的测试套件将 pyturso 与标准库sqlite3进行了大量逐项对比验证例如 tests/test_database.py 中同一组 SQL 在两种驱动下执行并断言结果完全一致如users [(1, alice), (2, bob)]。测试目录覆盖test_database.py同步 DB-API 驱动与sqlite3对比test_database_aio.py异步驱动test_database_sync.py同步驱动test_database_sync_aio.py异步同步驱动test_interrupt.py中断与查询超时行为test_sqlalchemy.py 与 test_sqlalchemy_async.pySQLAlchemy 方言。这些测试不仅验证了功能正确性也是理解驱动行为边界的绝佳参考——例如批量执行、参数绑定、异常映射等细节均有对应的断言用例。许可证与支持pyturso 以 MIT 许可证开源详见仓库根目录的 LICENSE.md。仓库内还提供了更多语言绑定的用法示例与 API 参考本文章节对应的全部 Python 源码、示例与测试均可从 bindings/python 目录继续深入阅读。【免费下载链接】tursoA SQL database in Rust: SQLite-compatible, now also speaking Postgres (experimental). The LLVM of databases.项目地址: https://gitcode.com/GitHub_Trending/tu/turso创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考