### Defining STRICT Table in SQLAlchemy Source: https://docs.sqlalchemy.org/en/20/dialects/sqlite.html/index This example shows how to define a SQLite table as `STRICT` using SQLAlchemy. This is accomplished by setting the `sqlite_strict` parameter to `True` when defining the `Table` object. This option was added in version 2.0.37. ```python Table("some_table", metadata, ..., sqlite_strict=True) ``` -------------------------------- ### Accessing Dotted Column Names with sqlite_raw_colnames Source: https://docs.sqlalchemy.org/en/20/dialects/sqlite.html/index This example shows how to bypass SQLAlchemy's default filtering of dotted column names by using the `sqlite_raw_colnames=True` execution option. This is useful when direct access to original, unmodified column names (including dots) is required, but it may affect Core and ORM queries using UNION. ```python result = conn.execution_options(sqlite_raw_colnames=True).exec_driver_sql( """ select x.a, x.b from x where a=1 union select x.a, x.b from x where a=2 """ ) assert result.keys() == ["x.a", "x.b"] ``` -------------------------------- ### SQLite Connect Strings Source: https://docs.sqlalchemy.org/en/20/dialects/sqlite.html/index Demonstrates various ways to construct connection strings for SQLite databases using SQLAlchemy's create_engine. Covers relative paths, absolute paths, and in-memory databases. ```python # relative path e = create_engine("sqlite:///path/to/database.db") # absolute path e = create_engine("sqlite:////path/to/database.db") # absolute path on Windows e = create_engine("sqlite:///C:\\path\\to\\database.db") # in-memory database (note three slashes) e = create_engine("sqlite:///:memory:") # also in-memory database e2 = create_engine("sqlite://") ``` -------------------------------- ### SQLite URI Connections with Driver Arguments Source: https://docs.sqlalchemy.org/en/20/dialects/sqlite.html/index Illustrates how to use SQLite's URI connection format with SQLAlchemy, allowing for driver-level arguments to be passed. This includes options like read-only mode and custom timeouts. ```python e = create_engine("sqlite:///file:path/to/database?mode=ro&uri=true") e = create_engine( "sqlite:///file:path/to/database?" "check_same_thread=true&timeout=10&mode=ro&nolock=1&uri=true" ) ``` -------------------------------- ### Connect to SQLite with aiosqlite (Asyncio) Source: https://docs.sqlalchemy.org/en/20/dialects/sqlite.html/index Establishes an asynchronous connection to a SQLite database using the aiosqlite dialect. This is useful for testing and prototyping with SQLAlchemy's asyncio interface. The connection string format is 'sqlite+aiosqlite:///file_path'. Arguments are passed directly to the pysqlite driver. ```python from sqlalchemy.ext.asyncio import create_async_engine engine = create_async_engine("sqlite+aiosqlite:///filename") ``` -------------------------------- ### SQLAlchemy Emits BEGIN for SQLite Transactions (sqlite3, aiosqlite) Source: https://docs.sqlalchemy.org/en/20/dialects/sqlite.html/index This configuration uses SQLAlchemy's event hooks to manage transaction control. The `PoolEvents.connect` hook disables SQLite's default transaction emission by setting `isolation_level` to `None`, while the `ConnectionEvents.begin` hook explicitly emits a 'BEGIN' command. This ensures consistent transaction behavior across different Python versions and drivers. ```python from sqlalchemy import create_engine, event engine = create_engine("sqlite:///myfile.db") @event.listens_for(engine, "connect") def do_connect(dbapi_connection, connection_record): # disable sqlite3's emitting of the BEGIN statement entirely. dbapi_connection.isolation_level = None @event.listens_for(engine, "begin") def do_begin(conn): # emit our own BEGIN. sqlite3 still emits COMMIT/ROLLBACK correctly conn.exec_driver_sql("BEGIN") ``` ```python from sqlalchemy import create_engine, event from sqlalchemy.ext.asyncio import create_async_engine engine = create_async_engine("sqlite+aiosqlite:///myfile.db") @event.listens_for(engine.sync_engine, "connect") def do_connect(dbapi_connection, connection_record): # disable aiosqlite's emitting of the BEGIN statement entirely. dbapi_connection.isolation_level = None @event.listens_for(engine.sync_engine, "begin") def do_begin(conn): # emit our own BEGIN. aiosqlite still emits COMMIT/ROLLBACK correctly conn.exec_driver_sql("BEGIN") ``` -------------------------------- ### SQLite Partial Index Creation (Python) Source: https://docs.sqlalchemy.org/en/20/dialects/sqlite.html/index Demonstrates how to create a partial index in SQLite using SQLAlchemy by specifying the `sqlite_where` argument within the `Index` constructor. This allows indexing only a subset of rows based on a condition. It requires SQLAlchemy's `Table`, `Column`, `Index`, and `and_` constructs. ```python from sqlalchemy import Table, Column, Integer, Index, and_ # Assuming m is a MetaData object # Example: # from sqlalchemy import MetaData # m = MetaData() tbl = Table("testtbl", m, Column("data", Integer)) idx = Index( "test_idx1", tbl.c.data, sqlite_where=and_(tbl.c.data > 5, tbl.c.data < 10), ) # The rendered SQL for the index creation would be: # CREATE INDEX test_idx1 ON testtbl (data) # WHERE data > 5 AND data < 10 ``` -------------------------------- ### Using a Custom DBAPI Module with pysqlcipher Source: https://docs.sqlalchemy.org/en/20/dialects/sqlite.html/index Demonstrates how to explicitly provide a DBAPI module for the pysqlcipher dialect when the default driver selection logic is not sufficient or a specific compatible driver is preferred. This is useful for ensuring compatibility with different SQLCipher driver implementations. ```python import sqlcipher_compatible_driver from sqlalchemy import create_engine e = create_engine( "sqlite+pysqlcipher://:password@/dbname.db", module=sqlcipher_compatible_driver, ) ``` -------------------------------- ### SQLAlchemy INSERT/UPDATE/DELETE RETURNING with SQLite Source: https://docs.sqlalchemy.org/en/20/dialects/sqlite.html/index Demonstrates how to use the RETURNING clause with INSERT, UPDATE, and DELETE statements in SQLAlchemy for the SQLite dialect. This allows fetching newly generated identifiers or updated/deleted row data. The `_UpdateBase.returning()` method is used for explicit RETURNING clauses. ```python from sqlalchemy.engine import Engine from sqlalchemy import event # INSERT..RETURNING result = connection.execute( table.insert().values(name="foo").returning(table.c.col1, table.c.col2) ) print(result.all()) # UPDATE..RETURNING result = connection.execute( table.update() .where(table.c.name == "foo") .values(name="bar") .returning(table.c.col1, table.c.col2) ) print(result.all()) # DELETE..RETURNING result = connection.execute( table.delete() .where(table.c.name == "foo") .returning(table.c.col1, table.c.col2) ) print(result.all()) ``` -------------------------------- ### SQLAlchemy SQLite Upsert with ON CONFLICT DO UPDATE/DO NOTHING Source: https://docs.sqlalchemy.org/en/20/dialects/sqlite.html/index Demonstrates the basic usage of SQLAlchemy's SQLite-specific insert dialect to perform upsert operations. It shows how to construct statements for both 'DO UPDATE' and 'DO NOTHING' conflict resolutions, specifying the conflict target by index elements. ```python from sqlalchemy.dialects.sqlite import insert insert_stmt = insert(my_table).values( id="some_existing_id", data="inserted value" ) do_update_stmt = insert_stmt.on_conflict_do_update( index_elements=["id"], set_=dict(data="updated value") ) # print(do_update_stmt) will output the SQL do_nothing_stmt = insert_stmt.on_conflict_do_nothing(index_elements=["id"]) # print(do_nothing_stmt) will output the SQL ``` -------------------------------- ### Enabling SQLite Foreign Key Support via SQLAlchemy Events Source: https://docs.sqlalchemy.org/en/20/dialects/sqlite.html/index Configures SQLAlchemy to automatically enable foreign key constraint checking for SQLite connections. This is achieved using the `event.listens_for` decorator on the 'connect' event for `Engine`. It requires SQLite version 3.6.19+ and specific compilation flags, and manually emits `PRAGMA foreign_keys=ON` for each new connection. ```python from sqlalchemy.engine import Engine from sqlalchemy import event @event.listens_for(Engine, "connect") def set_sqlite_pragma(dbapi_connection, connection_record): # the sqlite3 driver will not set PRAGMA foreign_keys # if autocommit=False; set to True temporarily ac = dbapi_connection.autocommit dbapi_connection.autocommit = True cursor = dbapi_connection.cursor() cursor.execute("PRAGMA foreign_keys=ON") cursor.close() # restore previous autocommit setting dbapi_connection.autocommit = ac ``` -------------------------------- ### Enable Non-Legacy SQLite Transactions via `autocommit` Parameter (sqlite3, aiosqlite) Source: https://docs.sqlalchemy.org/en/20/dialects/sqlite.html/index Configure SQLite to use modern transaction control by setting the `autocommit` parameter to `False` within `create_engine`'s `connect_args`. This approach works for both the synchronous `sqlite3` driver and the asynchronous `aiosqlite` driver. It ensures that transactions are managed explicitly, rather than relying on SQLite's default behavior. ```python from sqlalchemy import create_engine engine = create_engine( "sqlite:///myfile.db", connect_args={"autocommit": False} ) ``` ```python from sqlalchemy.ext.asyncio import create_async_engine engine = create_async_engine( "sqlite+aiosqlite:///myfile.db", connect_args={"autocommit": False} ) ``` -------------------------------- ### Illustrate SQLite Driver Bug with Dotted Column Names Source: https://docs.sqlalchemy.org/en/20/dialects/sqlite.html/index This code demonstrates a bug in SQLite versions prior to 3.10.0 where column names in result sets are incorrectly reported when using UNION. It sets up an in-memory SQLite database, creates a table, inserts data, and queries it, asserting the expected column names. ```python import sqlite3 assert sqlite3.sqlite_version_info < ( 3, 10, 0, ), "bug is fixed in this version" conn = sqlite3.connect(":memory:") cursor = conn.cursor() cursor.execute("create table x (a integer, b integer)") cursor.execute("insert into x (a, b) values (1, 1)") cursor.execute("insert into x (a, b) values (2, 2)") cursor.execute("select x.a, x.b from x") assert [c[0] for c in cursor.description] == ["a", "b"] cursor.execute( """ select x.a, x.b from x where a=1 union select x.a, x.b from x where a=2 """ ) assert [c[0] for c in cursor.description] == ["a", "b"], [ c[0] for c in cursor.description ] ``` -------------------------------- ### Enable Driver-Level Autocommit with `sqlite3` Source: https://docs.sqlalchemy.org/en/20/dialects/sqlite.html/index This code snippet shows how to enable SQLAlchemy's driver-level autocommit feature for the `sqlite3` driver. It involves creating an engine and setting the `isolation_level` parameter to 'AUTOCOMMIT'. This mode is incompatible with non-legacy transaction hooks and requires disabling explicit 'BEGIN' statements. ```python eng = create_engine("sqlite:///myfile.db", isolation_level="AUTOCOMMIT") ``` -------------------------------- ### Construct SQLite INSERT Statement Source: https://docs.sqlalchemy.org/en/20/dialects/sqlite.html/index Creates a SQLite-specific INSERT construct. This is a variant of the generic SQLAlchemy INSERT construct and includes methods for SQLite's ON CONFLICT syntax. ```python from sqlalchemy.dialects.sqlite import insert # Assuming 'my_table' is a SQLAlchemy Table object stmt = insert(my_table).values(name='value') print(stmt) ``` -------------------------------- ### Adding SQLite ON CONFLICT to Primary Key Constraint Source: https://docs.sqlalchemy.org/en/20/dialects/sqlite.html/index Demonstrates how to apply the FAIL conflict resolution algorithm to a PRIMARY KEY constraint using the `sqlite_on_conflict_primary_key` parameter on a `Column`. SQLAlchemy renders the PRIMARY KEY constraint separately, applying the conflict resolution to the constraint itself. ```Python from sqlalchemy import Table, Column, Integer, MetaData metadata = MetaData() some_table = Table( "some_table", metadata, Column( "id", Integer, primary_key=True, sqlite_on_conflict_primary_key="FAIL", ), ) # Renders CREATE TABLE some_table ( # id INTEGER NOT NULL, # PRIMARY KEY (id) ON CONFLICT FAIL # ) ``` -------------------------------- ### Use Temporary Tables with SQLAlchemy SQLite Source: https://docs.sqlalchemy.org/en/20/dialects/sqlite.html/index Demonstrates how to configure SQLAlchemy with SQLite to maintain temporary tables across connection pool checkouts. It shows using `SingletonThreadPool` for thread-local scope and `StaticPool` for multi-threaded scope. ```python # maintain the same connection per thread from sqlalchemy.pool import SingletonThreadPool from sqlalchemy import create_engine engine = create_engine("sqlite:///mydb.db", poolclass=SingletonThreadPool) ``` ```python # maintain the same connection across all threads from sqlalchemy.pool import StaticPool from sqlalchemy import create_engine engine = create_engine("sqlite:///mydb.db", poolclass=StaticPool) ``` -------------------------------- ### SQLite TIME Type Customization Source: https://docs.sqlalchemy.org/en/20/dialects/sqlite.html/index Illustrates how to customize the storage and parsing formats for the SQLite TIME type using `storage_format` and `regexp` parameters. This allows for non-standard time representations. The default uses `time.fromisoformat()` for parsing. ```python import re from sqlalchemy.dialects.sqlite import TIME t = TIME( storage_format="%(hour)02d-%(minute)02d-%(second)02d-%(microsecond)06d", regexp=re.compile("(\d+)-(\d+)-(\d+)-(?:-(\d+))?"), ) ``` -------------------------------- ### Customize DATE Storage Format and Parsing Source: https://docs.sqlalchemy.org/en/20/dialects/sqlite.html/index Configure SQLite DATE type with custom storage format string and regular expression for parsing incoming date strings. Supports named groups in regex for keyword argument mapping or positional groups for positional argument mapping to the date constructor. ```python import re from sqlalchemy.dialects.sqlite import DATE d = DATE( storage_format="%(month)02d/%(day)02d/%(year)04d", regexp=re.compile("(?P\d+)/(?P\d+)/(?P\d+)"), ) ``` -------------------------------- ### Adding SQLite ON CONFLICT to NOT NULL Constraint Source: https://docs.sqlalchemy.org/en/20/dialects/sqlite.html/index Illustrates how to specify the FAIL conflict resolution algorithm for a NOT NULL constraint using the `sqlite_on_conflict_not_null` parameter on a `Column`. This renders the column definition with an inline ON CONFLICT phrase. ```Python from sqlalchemy import Table, Column, Integer, MetaData metadata = MetaData() some_table = Table( "some_table", metadata, Column("id", Integer, primary_key=True), Column( "data", Integer, nullable=False, sqlite_on_conflict_not_null="FAIL" ), ) # Renders CREATE TABLE some_table ( # id INTEGER NOT NULL, # data INTEGER NOT NULL ON CONFLICT FAIL, # PRIMARY KEY (id) # ) ``` -------------------------------- ### Enable Non-Legacy SQLite Transactions via `PoolEvents.connect` (sqlite3 only) Source: https://docs.sqlalchemy.org/en/20/dialects/sqlite.html/index This snippet demonstrates how to enable non-legacy SQLite transaction control using the `PoolEvents.connect` event hook to set the `autocommit` attribute directly on the DBAPI connection. This method is specifically for the `sqlite3` driver and will not work with `aiosqlite` as it does not expose the `autocommit` attribute. ```python from sqlalchemy import create_engine, event engine = create_engine("sqlite:///myfile.db") @event.listens_for(engine, "connect") def do_connect(dbapi_connection, connection_record): # enable autocommit=False mode dbapi_connection.autocommit = False ``` -------------------------------- ### Construct SQLite JSON Type Source: https://docs.sqlalchemy.org/en/20/dialects/sqlite.html/index Create a JSON type instance for SQLite with optional none_as_null parameter. The JSON type inherits from sqlalchemy.types.JSON and adapts JSON operations to use JSON_EXTRACT and JSON_QUOTE functions at the database level. ```python from sqlalchemy.dialects.sqlite import JSON json_type = JSON(none_as_null=False) ``` -------------------------------- ### Connect to SQLite with pysqlcipher (Encrypted) Source: https://docs.sqlalchemy.org/en/20/dialects/sqlite.html/index Connects to an encrypted SQLite database using the pysqlcipher dialect. It supports SQLCipher backends and allows specifying a passphrase and other encryption-related PRAGMA commands in the connection URL. The format is 'sqlite+pysqlcipher://:passphrase@/file_path[?kdf_iter=]'. ```python from sqlalchemy import create_engine e = create_engine("sqlite+pysqlcipher://:testing@/foo.db") ``` ```python e = create_engine("sqlite+pysqlcipher://:testing@//path/to/foo.db") ``` ```python e = create_engine( "sqlite+pysqlcipher://:testing@/foo.db?cipher=aes-256-cfb&kdf_iter=64000" ) ``` -------------------------------- ### Register User-Defined Functions (UDFs) with SQLAlchemy SQLite Source: https://docs.sqlalchemy.org/en/20/dialects/sqlite.html/index Shows how to register Python user-defined functions with an SQLite database connection in SQLAlchemy using event listeners. The `create_function` method is called on the DBAPI connection when it's established. ```python from sqlalchemy import create_engine from sqlalchemy import event from sqlalchemy import text def udf(): return "udf-ok" engine = create_engine("sqlite:///./db_file") @event.listens_for(engine, "connect") def connect(conn, rec): conn.create_function("udf", 0, udf) for i in range(5): with engine.connect() as conn: print(conn.scalar(text("SELECT UDF()!)")))) ``` -------------------------------- ### Adding SQLite ON CONFLICT to UniqueConstraint Source: https://docs.sqlalchemy.org/en/20/dialects/sqlite.html/index Demonstrates how to add a UNIQUE constraint with the IGNORE conflict resolution algorithm using the `sqlite_on_conflict` parameter on a `UniqueConstraint` object. This renders a UNIQUE constraint with the ON CONFLICT clause in the generated DDL. ```Python from sqlalchemy import Table, Column, Integer, MetaData, UniqueConstraint metadata = MetaData() some_table = Table( "some_table", metadata, Column("id", Integer, primary_key=True), Column("data", Integer), UniqueConstraint("id", "data", sqlite_on_conflict="IGNORE"), ) # Renders CREATE TABLE some_table ( # id INTEGER NOT NULL, # data INTEGER, # PRIMARY KEY (id), # UNIQUE (id, data) ON CONFLICT IGNORE # ) ``` -------------------------------- ### Configuring Engine with sqlite_raw_colnames Source: https://docs.sqlalchemy.org/en/20/dialects/sqlite.html/index This snippet demonstrates how to set the `sqlite_raw_colnames` execution option globally for an entire SQLAlchemy engine. This ensures that all subsequent connections and operations using this engine will return raw column names, including any dots. ```python engine = create_engine( "sqlite://", execution_options={"sqlite_raw_colnames": True} ) ``` -------------------------------- ### Including Internal SQLite Schema Tables during Reflection Source: https://docs.sqlalchemy.org/en/20/dialects/sqlite.html/index This code demonstrates how to use the `sqlite_include_internal=True` parameter with SQLAlchemy's reflection methods, such as `MetaData.reflect()` or `Inspector.get_table_names()`. This allows the retrieval of SQLite internal schema objects (prefixed with `sqlite_`), which are normally omitted. ```python # Example for MetaData.reflect() metadata.reflect(bind=engine, sqlite_include_internal=True) # Example for Inspector.get_table_names() inspector.get_table_names(include_internal=True) ``` -------------------------------- ### Create Regular Expression Function in SQLite using pysqlite Source: https://docs.sqlalchemy.org/en/20/dialects/sqlite.html/index Implements regular expression support for SQLite using Python's re.search function through pysqlite's create_function hook. SQLite lacks native regex support, requiring a user-defined function to enable the REGEXP operator. Note that regex flags are not separately supported but can be included inline within the pattern string. ```python def regexp(a, b): return re.search(a, b) is not None sqlite_connection.create_function( "regexp", 2, regexp, ) ``` -------------------------------- ### Configure Native DateTime Support in SQLite Engine Source: https://docs.sqlalchemy.org/en/20/dialects/sqlite.html/index Configures SQLAlchemy's create_engine to enable pysqlite's native date/datetime type parsing using PARSE_DECLTYPES and PARSE_COLNAMES options. When native_datetime=True is set, DATE and TIMESTAMP types bypass SQLAlchemy's bind parameter and result processing, though this configuration is generally not recommended for standard SQLAlchemy usage. ```python engine = create_engine( "sqlite://", connect_args={ "detect_types": sqlite3.PARSE_DECLTYPES | sqlite3.PARSE_COLNAMES }, native_datetime=True, ) ``` -------------------------------- ### SQLAlchemy Workaround for Dotted Column Names Bug Source: https://docs.sqlalchemy.org/en/20/dialects/sqlite.html/index This Python code demonstrates how SQLAlchemy's dialect handles the SQLite bug with dotted column names. It uses SQLAlchemy's `create_engine` and `exec_driver_sql` to execute queries, showing that SQLAlchemy filters out the dots in column names by default, ensuring predictable results for `result.keys()`. ```python from sqlalchemy import create_engine eng = create_engine("sqlite://") conn = eng.connect() conn.exec_driver_sql("create table x (a integer, b integer)") conn.exec_driver_sql("insert into x (a, b) values (1, 1)") conn.exec_driver_sql("insert into x (a, b) values (2, 2)") result = conn.exec_driver_sql("select x.a, x.b from x") assert result.keys() == ["a", "b"] result = conn.exec_driver_sql( """ select x.a, x.b from x where a=1 union select x.a, x.b from x where a=2 """ ) assert result.keys() == ["a", "b"] ``` -------------------------------- ### Adding SQLite ON CONFLICT to Column Unique Constraint Source: https://docs.sqlalchemy.org/en/20/dialects/sqlite.html/index Shows how to apply the IGNORE conflict resolution algorithm to a UNIQUE constraint defined directly on a `Column` using the `sqlite_on_conflict_unique` parameter. This renders a UNIQUE constraint on the specified column with the ON CONFLICT clause. ```Python from sqlalchemy import Table, Column, Integer, MetaData metadata = MetaData() some_table = Table( "some_table", metadata, Column("id", Integer, primary_key=True), Column( "data", Integer, unique=True, sqlite_on_conflict_unique="IGNORE" ), ) # Renders CREATE TABLE some_table ( # id INTEGER NOT NULL, # data INTEGER, # PRIMARY KEY (id), # UNIQUE (data) ON CONFLICT IGNORE # ) ``` -------------------------------- ### SQLAlchemy SQLite Upsert with SET Clause for Updates Source: https://docs.sqlalchemy.org/en/20/dialects/sqlite.html/index Shows how to define the update action for an ON CONFLICT...DO UPDATE clause in SQLAlchemy for SQLite. The 'set_' parameter accepts a dictionary to specify direct values for updating existing rows. ```python stmt = insert(my_table).values(id="some_id", data="inserted value") do_update_stmt = stmt.on_conflict_do_update( index_elements=["id"], set_=dict(data="updated value") ) # print(do_update_stmt) will output the SQL ``` -------------------------------- ### SQLAlchemy SQLite Upsert with Partial Index Target Source: https://docs.sqlalchemy.org/en/20/dialects/sqlite.html/index Illustrates how to specify a partial index as the conflict target for an upsert operation in SQLAlchemy for SQLite. This allows the ON CONFLICT clause to be applied only when specific conditions are met, demonstrated using the 'index_where' parameter. ```python stmt = insert(my_table).values(user_email="a@b.com", data="inserted data") do_update_stmt = stmt.on_conflict_do_update( index_elements=[my_table.c.user_email], index_where=my_table.c.user_email.like("%@gmail.com"), set_=dict(data=stmt.excluded.data), ) # print(do_update_stmt) will output the SQL ``` -------------------------------- ### Custom Mixed Binary Type for SQLAlchemy SQLite Source: https://docs.sqlalchemy.org/en/20/dialects/sqlite.html/index Defines a custom SQLAlchemy `TypeDecorator` named `MixedBinary` to handle columns in SQLite that may contain mixed string and binary data. This type ensures values are consistently processed as bytes. ```python from sqlalchemy import String from sqlalchemy import TypeDecorator class MixedBinary(TypeDecorator): impl = String cache_ok = True def process_result_value(self, value, dialect): if isinstance(value, str): value = bytes(value, "utf-8") elif value is not None: value = bytes(value) return value ``` -------------------------------- ### SQLite INSERT with ON CONFLICT DO NOTHING Source: https://docs.sqlalchemy.org/en/20/dialects/sqlite.html/index Specifies a DO NOTHING action for the ON CONFLICT clause in a SQLite INSERT statement. This is useful for avoiding errors when attempting to insert duplicate unique keys. ```python from sqlalchemy.dialects.sqlite import insert from sqlalchemy import Table, Column, Integer, String # Assuming 'my_table' is defined with a unique constraint my_table = Table('my_table', metadata, Column('id', Integer, primary_key=True), Column('name', String, unique=True)) stmt = insert(my_table).values(name='unique_value') stmt = stmt.on_conflict_do_nothing() print(stmt) ``` -------------------------------- ### Customize DATETIME Storage Format and Parsing Source: https://docs.sqlalchemy.org/en/20/dialects/sqlite.html/index Configure SQLite DATETIME type with custom storage format string and regular expression for parsing incoming datetime strings. Supports optional microsecond truncation. The DATETIME class inherits from sqlalchemy.dialects.sqlite.base._DateTimeMixin and sqlalchemy.types.DateTime. ```python import re from sqlalchemy.dialects.sqlite import DATETIME dt = DATETIME( storage_format=( "%(year)04d/%(month)02d/%(day)02d %(hour)02d:%(minute)02d:%(second)02d" ), regexp=r"(\d+)/(\d+)/(\d+) (\d+)-(\d+)-(\d+)", ) ``` -------------------------------- ### SQLite INSERT ON CONFLICT with WHERE Clause (Python) Source: https://docs.sqlalchemy.org/en/20/dialects/sqlite.html/index Demonstrates how to use `Insert.on_conflict_do_update()` with a `where` clause to conditionally update rows on conflict. This method limits updates to rows matching the specified WHERE condition. It requires SQLAlchemy and a defined table object. ```python from sqlalchemy import insert, Table, Column, Integer # Assuming my_table is a SQLAlchemy Table object # Example: # metadata = MetaData() # my_table = Table( # "my_table", metadata, # Column("id", Integer, primary_key=True), # Column("data", String), # Column("author", String), # Column("status", Integer) # ) stmt = insert(my_table).values( id="some_id", data="inserted value", author="jlh" ) on_update_stmt = stmt.on_conflict_do_update( index_elements=["id"], set_=dict(data="updated value", author=stmt.excluded.author), where=(my_table.c.status == 2), ) print(on_update_stmt) ``` -------------------------------- ### Configure Autoincrement for Non-Integer Types in SQLAlchemy SQLite Source: https://docs.sqlalchemy.org/en/20/dialects/sqlite.html/index Shows how to ensure autoincrement behavior for primary keys using types other than standard Integer in SQLite. It covers using `TypeEngine.with_variant()` to map a type to INTEGER specifically for SQLite, and creating custom type subclasses with compilation overrides. ```python from sqlalchemy import Table, Column, MetaData, BigInteger, Integer metadata = MetaData() # Using with_variant table_with_variant = Table( "my_table_variant", metadata, Column( "id", BigInteger().with_variant(Integer, "sqlite"), primary_key=True, ), ) # Using custom type subclass with compilation override from sqlalchemy.ext.compiler import compiles class SLBigInteger(BigInteger): pass @compiles(SLBigInteger, "sqlite") def bi_c_sqlite(element, compiler, **kw): return "INTEGER" @compiles(SLBigInteger) def bi_c_default(element, compiler, **kw): return compiler.visit_BIGINT(element, **kw) table_custom_type = Table( "my_table_custom", metadata, Column("id", SLBigInteger(), primary_key=True) ) ``` -------------------------------- ### Import SQLite Data Types from SQLAlchemy Dialect Source: https://docs.sqlalchemy.org/en/20/dialects/sqlite.html/index Import all UPPERCASE SQLite-compatible data types from the sqlalchemy.dialects.sqlite module. These types are valid for SQLite and can be used in column definitions. ```python from sqlalchemy.dialects.sqlite import ( BLOB, BOOLEAN, CHAR, DATE, DATETIME, DECIMAL, FLOAT, INTEGER, NUMERIC, JSON, SMALLINT, TEXT, TIME, TIMESTAMP, VARCHAR, ) ``` -------------------------------- ### Defining Table WITHOUT ROWID in SQLAlchemy Source: https://docs.sqlalchemy.org/en/20/dialects/sqlite.html/index This code illustrates how to define a SQLite table using SQLAlchemy's `Table` construct with the `WITHOUT ROWID` option. This is achieved by setting the `sqlite_with_rowid` parameter to `False` during table creation. ```python Table("some_table", metadata, ..., sqlite_with_rowid=False) ``` -------------------------------- ### Disable Connection Pooling for SQLite File Databases Source: https://docs.sqlalchemy.org/en/20/dialects/sqlite.html/index Disables connection pooling for file-based SQLite databases by specifying NullPool as the poolclass parameter. Use this approach when experiencing file locking issues, though it incurs a small performance overhead compared to QueuePool due to lack of connection reuse. ```python from sqlalchemy import NullPool engine = create_engine("sqlite:///myfile.db", poolclass=NullPool) ``` -------------------------------- ### SQLite INSERT ON CONFLICT DO NOTHING (Python) Source: https://docs.sqlalchemy.org/en/20/dialects/sqlite.html/index Illustrates using `Insert.on_conflict_do_nothing()` to skip inserting a row entirely if a unique constraint violation occurs. This is useful for de-duplicating data or ensuring idempotency. It requires SQLAlchemy and a defined table object. ```python from sqlalchemy import insert, Table, Column, Integer # Assuming my_table is a SQLAlchemy Table object # Example: # metadata = MetaData() # my_table = Table( # "my_table", metadata, # Column("id", Integer, primary_key=True), # Column("data", String) # ) stmt = insert(my_table).values(id="some_id", data="inserted value") stmt = stmt.on_conflict_do_nothing(index_elements=["id"]) print(stmt) ``` ```python from sqlalchemy import insert, Table, Column, Integer # Assuming my_table is a SQLAlchemy Table object # Example: # metadata = MetaData() # my_table = Table( # "my_table", metadata, # Column("id", Integer, primary_key=True), # Column("data", String) # ) stmt = insert(my_table).values(id="some_id", data="inserted value") stmt = stmt.on_conflict_do_nothing() print(stmt) ``` -------------------------------- ### Use Memory Database in Multiple Threads with SQLAlchemy SQLite Source: https://docs.sqlalchemy.org/en/20/dialects/sqlite.html/index Configures SQLAlchemy to use an in-memory SQLite database in a multithreaded environment by sharing a single connection object. It utilizes `StaticPool` and disables the `check_same_thread` flag for the Pysqlite driver. ```python from sqlalchemy.pool import StaticPool from sqlalchemy import create_engine engine = create_engine( "sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool, ) ``` -------------------------------- ### Enable AUTOINCREMENT Keyword for Primary Key in SQLAlchemy SQLite Source: https://docs.sqlalchemy.org/en/20/dialects/sqlite.html/index Demonstrates how to explicitly render the AUTOINCREMENT keyword on a primary key column in a SQLAlchemy Table definition for SQLite. This is achieved by setting the `sqlite_autoincrement=True` flag on the Table construct. ```python from sqlalchemy import Table, Column, Integer, MetaData metadata = MetaData() Table( "sometable", metadata, Column("id", Integer, primary_key=True), sqlite_autoincrement=True, ) ``` -------------------------------- ### SQLite INSERT with ON CONFLICT DO UPDATE Source: https://docs.sqlalchemy.org/en/20/dialects/sqlite.html/index Specifies a DO UPDATE SET action for the ON CONFLICT clause in a SQLite INSERT statement. This allows for updating existing rows when a conflict occurs. ```python from sqlalchemy.dialects.sqlite import insert from sqlalchemy import Table, Column, Integer, String, bindparam # Assuming 'my_table' is defined with a unique constraint my_table = Table('my_table', metadata, Column('id', Integer, primary_key=True), Column('name', String, unique=True), Column('count', Integer)) # Using bindparam for values to update stmt = insert(my_table).values(name='unique_value', count=1) stmt = stmt.on_conflict_do_update(set_=dict(count=my_table.c.count + bindparam('new_count'))) print(stmt) ``` -------------------------------- ### SQLAlchemy SQLite Upsert Using Excluded INSERT Values Source: https://docs.sqlalchemy.org/en/20/dialects/sqlite.html/index Demonstrates how to reference values from the proposed insert statement within the ON CONFLICT...DO UPDATE clause using the 'Insert.excluded' alias. This allows updating existing rows with values that would have been inserted. ```python stmt = insert(my_table).values( id="some_id", data="inserted value", author="jlh" ) do_update_stmt = stmt.on_conflict_do_update( index_elements=["id"], set_=dict(data="updated value", author=stmt.excluded.author), ) # print(do_update_stmt) will output the SQL ``` -------------------------------- ### Accessing Excluded Namespace in ON CONFLICT Source: https://docs.sqlalchemy.org/en/20/dialects/sqlite.html/index Provides access to the 'excluded' namespace within a SQLite ON CONFLICT statement. This allows referencing the row that would have been inserted. ```python from sqlalchemy.dialects.sqlite import insert from sqlalchemy import Table, Column, Integer, String # Assuming 'my_table' is defined with a unique constraint my_table = Table('my_table', metadata, Column('id', Integer, primary_key=True), Column('name', String, unique=True), Column('description', String)) stmt = insert(my_table).values(name='existing_value', description='new desc') stmt = stmt.on_conflict_do_update(set_=dict(description=insert(my_table).excluded.description)) print(stmt) ``` -------------------------------- ### Persist None as SQL NULL in SQLite JSON Source: https://docs.sqlalchemy.org/en/20/dialects/sqlite.html/index Demonstrates how to configure SQLAlchemy to persist Python's `None` as SQL NULL instead of JSON's `null` for JSON type fields in SQLite. It shows the use of `sqlalchemy.null()` for explicit NULL insertion. Note that this setting does not affect `Column.default` or `Column.server_default`. ```python from sqlalchemy import null conn.execute(table.insert(), {"data": null()}) ``` === COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.