Skip to content

SQLALCHEMY_DIALECT.md

Latest commit

 

History

History
281 lines (206 loc) · 10.5 KB

SQLALCHEMY_DIALECT.md

File metadata and controls

281 lines (206 loc) · 10.5 KB
Raw

SQLAlchemy Dialect for Pyturso

This document describes the SQLAlchemy dialect implementation for pyturso.

Status: Implemented

The SQLAlchemy dialect is implemented with three dialects:

  • sqlite+turso:// - Basic local database connections
  • sqlite+aioturso:// - Basic local database connections for SQLAlchemy async engines
  • sqlite+turso_sync:// - Sync-enabled connections with remote database support

Installation

Requires SQLAlchemy ≥ 2.0.45

pip install pyturso[sqlalchemy] # ensures compatible version of SQLAlchemy is installed

Quick Start

Basic Local Connection

from sqlalchemy import create_engine, text

# In-memory database
engine = create_engine("sqlite+turso:///:memory:")

# File-based database
engine = create_engine("sqlite+turso:///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)

Sync-Enabled Connection (Remote Sync)

from sqlalchemy import create_engine, text
from turso.sqlalchemy import get_sync_connection

# Via URL query parameters
engine = create_engine(
    "sqlite+turso_sync:///local.db"
    "?remote_url=https://your-db.turso.io"
    "&auth_token=your-token"
)

# Or via connect_args (supports callables for dynamic tokens)
engine = create_engine(
    "sqlite+turso_sync:///local.db",
    connect_args={
        "remote_url": "https://your-db.turso.io",
        "auth_token": lambda: get_fresh_token(),
    }
)

with engine.connect() as conn:
    sync = get_sync_connection(conn) # get_sync_connection() 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 remote

Async Local Connection

from sqlalchemy import text
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine

engine = create_async_engine("sqlite+aioturso:///:memory:")

async with engine.begin() as conn:
    await conn.execute(text("CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT)"))
    await conn.execute(text("INSERT INTO users (name) VALUES ('Alice')"))

async with AsyncSession(engine) as session:
    result = await session.execute(text("SELECT name FROM users ORDER BY id"))
    print(result.scalars().all())

await engine.dispose()

ORM Usage

from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import declarative_base, Session

Base = declarative_base()

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

engine = create_engine("sqlite+turso:///:memory:")
Base.metadata.create_all(engine)

with Session(engine) as session:
    session.add(User(name="Alice"))
    session.commit()

    users = session.query(User).all()

URL Formats

Basic Dialect (sqlite+turso://)

sqlite+turso:///path/to/database.db
sqlite+turso:///:memory:
sqlite+turso:///db.db?isolation_level=IMMEDIATE

Query parameters:

  • isolation_level - Transaction isolation level: DEFERRED (default), IMMEDIATE, EXCLUSIVE, or AUTOCOMMIT (disables implicit transactions)
  • experimental_features - Comma-separated feature flags

Async Local Dialect (sqlite+aioturso://)

sqlite+aioturso:///path/to/database.db
sqlite+aioturso:///:memory:
sqlite+aioturso:///db.db?isolation_level=IMMEDIATE

Query parameters:

  • isolation_level - Transaction isolation level: DEFERRED (default), IMMEDIATE, EXCLUSIVE, or AUTOCOMMIT (disables implicit transactions)
  • experimental_features - Comma-separated feature flags

Sync Dialect (sqlite+turso_sync://)

sqlite+turso_sync:///local.db?remote_url=https://db.turso.io&auth_token=xxx

Query parameters:

  • remote_url (required) - Remote Turso/libsql server URL
  • auth_token - Authentication token
  • client_name - Client identifier (default: turso-sqlalchemy)
  • long_poll_timeout_ms - Long poll timeout in milliseconds
  • bootstrap_if_empty - Bootstrap from remote if local empty (default: true)
  • isolation_level - Transaction isolation level: DEFERRED (default), IMMEDIATE, EXCLUSIVE, or AUTOCOMMIT (disables implicit transactions)
  • experimental_features - Comma-separated feature flags

URL validation:

  • Username/password in URL raises ValueError (use auth_token instead)
  • Host/port in URL raises ValueError (use remote_url query param instead)
  • Unrecognized query parameters emit a UserWarning

Sync Operations

The get_sync_connection() helper (shown in Quick Start above) exposes the underlying turso.sync.ConnectionSync with these sync-specific methods:

  • pull() - Pull changes from remote; returns True if updates were pulled
  • push() - Push local changes to remote
  • checkpoint() - Checkpoint the WAL
  • stats() - Sync statistics (e.g. stats().network_received_bytes)

get_sync_connection() raises TypeError if called on a non-sync connection (e.g. a plain sqlite+turso:// or standard sqlite:// engine).

Architecture

_TursoDialectMixin (reflection overrides)
        │
        │   SQLiteDialect_pysqlite (SQLAlchemy built-in)
        │           │
        ├───────────┤
        │           │
        ├── TursoDialect (sqlite+turso://)
        │       ├── uses turso.connect()
        │       └── pool: SingletonThreadPool (:memory:) / QueuePool (file)
        │
        ├── AioTursoDialect (sqlite+aioturso://)
        │       ├── uses turso.aio.connect()
        │       ├── adapts turso.aio to SQLAlchemy's DBAPI-shaped async contract
        │       └── pool: StaticPool (:memory:) / AsyncAdaptedQueuePool (file)
        │
        └── TursoSyncDialect (sqlite+turso_sync://)
                ├── uses turso.sync.connect()
                ├── pool: SingletonThreadPool (:memory:) / QueuePool (file)
                └── get_sync_connection() → ConnectionSync (pull/push/checkpoint/stats)

The sync dialects use Python MRO: _TursoDialectMixin provides PRAGMA-related overrides, SQLiteDialect_pysqlite provides core SQLite dialect behavior. The async dialect uses SQLiteDialect_aiosqlite with the same Turso-specific mixin and an adapter that maps turso.aio into SQLAlchemy's async DBAPI wrapper.

What Pyturso Provides

Requirement Status
apilevel = "2.0" Provided
threadsafety = 1 Provided
paramstyle = "qmark" Provided
sqlite_version Provided
sqlite_version_info Provided
connect() function Provided
Connection class Provided
Cursor class Provided
Exception hierarchy Provided

Both turso and turso.sync modules expose the full DB-API 2.0 interface including exception hierarchy (Warning, Error, InterfaceError, DatabaseError, DataError, OperationalError, IntegrityError, InternalError, ProgrammingError, NotSupportedError).

turso.aio exposes coroutine connection and cursor APIs, but it does not expose the DB-API module metadata and exception hierarchy directly. sqlite+aioturso:// uses an internal adapter to mirror those DB-API module attributes from turso and to provide SQLite constants such as PARSE_DECLTYPES, PARSE_COLNAMES, and Binary.

Dialect Overrides

All dialects share these overrides via _TursoDialectMixin and direct method implementations:

Class Attributes

  • supports_statement_cache = True - Enables SQLAlchemy statement caching for performance
  • supports_native_datetime = False - Turso handles datetime as strings, not native types

Method Overrides

  • import_dbapi() - Returns turso, turso.sync, or the async adapter for turso.aio
  • create_connect_args() - Parses URL to connection arguments
  • on_connect() - Returns None (skips REGEXP function setup that pysqlite does, since turso doesn't support create_function)
  • get_isolation_level() - Returns SERIALIZABLE (turso doesn't support PRAGMA read_uncommitted)
  • set_isolation_level() - No-op (isolation set at connection time via isolation_level param)
  • get_pool_class() - Returns SingletonThreadPool for sync :memory:, QueuePool for sync file databases, StaticPool for async :memory:, and AsyncAdaptedQueuePool for async file databases

Reflection Overrides (via _TursoDialectMixin)

Index, unique-constraint, check-constraint, and foreign-key reflection (get_indexes, get_unique_constraints, get_check_constraints, get_foreign_keys, and their get_multi_* counterparts) are inherited from SQLAlchemy's parent SQLite dialect — Turso supports PRAGMA index_list / index_info / index_xinfo / foreign_key_list and returns the original DDL via sqlite_master.sql.

The following are overridden and return empty stubs:

  • get_temp_table_names() / get_temp_view_names() - no temp database (sqlite_temp_master not supported)

Limitations

Native Datetime

supports_native_datetime is set to False. Datetime columns should use String type and store ISO format strings. SQLAlchemy's DateTime type will still work but values are stored/retrieved as strings.

Async Scope

sqlite+aioturso:// supports local databases through turso.aio. Remote sync for SQLAlchemy async engines is not implemented by this dialect; use sqlite+turso_sync:// with synchronous SQLAlchemy engines for remote sync operations.

Entry Points

Dialects are registered via pyproject.toml entry points:

[project.entry-points."sqlalchemy.dialects"]
"sqlite.turso" = "turso.sqlalchemy:TursoDialect"
"sqlite.aioturso" = "turso.sqlalchemy:AioTursoDialect"
"sqlite.turso_sync" = "turso.sqlalchemy:TursoSyncDialect"

Files

  • turso/sqlalchemy/__init__.py - Module exports (TursoDialect, AioTursoDialect, TursoSyncDialect, get_sync_connection)
  • turso/sqlalchemy/dialect.py - Dialect implementations, async DBAPI adapter, and _TursoDialectMixin
  • tests/test_sqlalchemy.py - Sync SQLAlchemy dialect tests
  • tests/test_sqlalchemy_async.py - Async SQLAlchemy dialect tests

References