===============
LIBRARY RULES
===============
From library maintainers:
- cysqlite operates in pure autocommit mode as this is how SQLite itself operates. Statements execute immediately unless you’ve explicitly opened a transaction.
- cysqlite provides a helpful `atomic()` decorator or context-manager that can be nested for managing transactions and savepoints.
- cysqlite provides an asyncio implementation `cysqlite.aio` which can replace `aiosqlite`.

### Install Cysqlite from Source

Source: https://github.com/coleifer/cysqlite/blob/master/_autodocs/overview.md

Install Cysqlite by compiling from source, which links against the system's SQLite library. Use this option if you prefer to manage your SQLite installation separately.

```bash
pip install --no-binary :all: cysqlite
```

--------------------------------

### Async Connection Example

Source: https://github.com/coleifer/cysqlite/blob/master/_autodocs/asyncio.md

Create an async database connection. Must be called within an event loop.

```python
import asyncio
from cysqlite.aio import connect

async def main():
    db = await connect(':memory:')
    await db.close()

asyncio.run(main())
```

--------------------------------

### Install Cysqlite with SQLCipher Support

Source: https://github.com/coleifer/cysqlite/blob/master/_autodocs/overview.md

Install Cysqlite with SQLCipher support by cloning the repository, fetching the SQLCipher source, and then installing the package. This enables encrypted SQLite databases.

```bash
git clone https://github.com/coleifer/cysqlite
cd cysqlite
./scripts/fetch_sqlcipher
SQLCIPHER=1 pip install .
```

--------------------------------

### Install Latest Commit from Git

Source: https://github.com/coleifer/cysqlite/blob/master/docs/installation.md

Install the latest development version of cysqlite directly from its GitHub repository.

```shell
# (note: links against system sqlite)
pip install -e git+https://github.com/coleifer/cysqlite.git#egg=cysqlite
```

--------------------------------

### Install Cysqlite with Embedded SQLite

Source: https://github.com/coleifer/cysqlite/blob/master/_autodocs/overview.md

Install Cysqlite using a pre-built wheel that includes an embedded SQLite library. This is the simplest installation method.

```bash
pip install cysqlite
```

--------------------------------

### Complete Async Example with Transactions and Pooling

Source: https://github.com/coleifer/cysqlite/blob/master/_autodocs/asyncio.md

Demonstrates setting up an in-memory database, performing operations within a transaction, querying data, and utilizing a connection pool for concurrent reads.

```python
import asyncio
from cysqlite.aio import connect, Pool

async def setup():
    db = await connect(':memory:')
    await db.execute('CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT, age INT)')
    return db

async def query_with_transaction():
    db = await setup()
    
    # Use transaction context manager
    async with db.transaction():
        await db.execute('INSERT INTO users (name, age) VALUES (?, ?)', ('Alice', 30))
        await db.execute('INSERT INTO users (name, age) VALUES (?, ?)', ('Bob', 25))
    
    # Query results
    cursor = await db.execute('SELECT * FROM users ORDER BY age')
    async for row in cursor:
        print(f'{row[1]}: {row[2]} years old')
    
    await db.close()

async def use_pool():
    pool = await Pool(':memory:', readers=2, writer=True)
    
    # Setup via writer
    async with pool.writer() as db:
        await db.execute('CREATE TABLE data (id INT PRIMARY KEY, value TEXT)')
        await db.execute('INSERT INTO data VALUES (1, "test")')
    
    # Concurrent reads
    async def read_data():
        async with pool.reader() as db:
            count = await db.execute_scalar('SELECT COUNT(*) FROM data')
            return count
    
    counts = await asyncio.gather(read_data(), read_data(), read_data())
    print(f'Read counts: {counts}')
    
    await pool.close()

async def main():
    await query_with_transaction()
    await use_pool()

asyncio.run(main())
```

--------------------------------

### Configure Database Settings

Source: https://github.com/coleifer/cysqlite/blob/master/docs/api.md

Demonstrates querying and setting database configuration options using `db_config`. This example shows how to query the foreign key setting and enable defensive mode.

```python
current = db.db_config(SQLITE_DBCONFIG_ENABLE_FKEY)
db.db_config(SQLITE_DBCONFIG_DEFENSIVE, 1)
```

--------------------------------

### Basic Transaction Usage

Source: https://github.com/coleifer/cysqlite/blob/master/docs/aio.md

Shows a simple example of using the `transaction` context manager to execute multiple SQL statements that will be committed together.

```python
async with db.transaction() as txn:
    await db.execute('insert into users (name) values (?)', ('alice',))
    await db.execute('insert into users (name) values (?)', ('bob',))

# Both rows committed.
```

--------------------------------

### Basic cysqlite usage example

Source: https://github.com/coleifer/cysqlite/blob/master/README.md

Demonstrates connecting to an in-memory database, creating a table, inserting data within a transaction, and querying results using both positional and named parameters.

```python
from cysqlite import connect

db = connect(':memory:')

db.execute('create table data (k, v)')

with db.atomic():
    db.executemany('insert into data (k, v) values (?, ?)',
                   [(f'k{i:02d}', f'v{i:02d}') for i in range(10)])
    print(db.last_insert_rowid())  # 10.

curs = db.execute('select * from data')
for row in curs:
    print(row)  # e.g., ('k00', 'v00')

# We can use named parameters with a dict as well.
row = db.execute_one('select * from data where k = :key and v = :val',
                     {'key': 'k05', 'val': 'v05'})
print(row)  # ('k05', 'v05')

db.close()
```

--------------------------------

### Pool Connection Management Example

Source: https://github.com/coleifer/cysqlite/blob/master/docs/utils.md

Demonstrates how to use the Pool class for managing read-only and read/write connections. Shows overriding pragmas and handling read-only errors.

```python
pool = Pool('app.db', pragmas={'cache_size': -128000})

with pool.writer() as conn:
    conn.execute('create table if not exists users ('
                 '"id" integer not null primary key, '
                 '"username" text not null)')

    conn.execute('insert into users (username) values (?)', ('alice',))

with pool.reader() as conn:
    curs = conn.execute('select * from users')

    # Raises OperationalError - connection is read-only.
    conn.execute('insert into users (username) values (?)', ('bob',))
```

--------------------------------

### Complete Type System Example with Custom Adapters and Converters

Source: https://github.com/coleifer/cysqlite/blob/master/_autodocs/types.md

Illustrates the full type system by registering custom adapters and converters for Decimal and date types, creating a table with these types, and then inserting and retrieving data.

```python
from cysqlite import connect
from cysqlite.metadata import Column, Index
from decimal import Decimal
from datetime import date

db = connect('/data/store.db')

# Register custom type adapters and converters
@db.adapter(Decimal)
def adapt_decimal(d):
    return float(d)

@db.converter('MONEY')
def convert_money(v):
    return Decimal(str(v))

@db.adapter(date)
def adapt_date(d):
    return d.isoformat()

@db.converter('DATE')
def convert_date(s):
    return date.fromisoformat(s)

# Create schema with custom types
db.execute('''
    CREATE TABLE IF NOT EXISTS products (
        id INTEGER PRIMARY KEY,
        sku TEXT NOT NULL UNIQUE,
        name TEXT NOT NULL,
        price MONEY NOT NULL,
        cost MONEY,
        added_date DATE NOT NULL,
        updated_at DATETIME DEFAULT CURRENT_TIMESTAMP
    )
''')

db.execute('CREATE INDEX idx_sku ON products(sku)')

# Introspect schema
columns = db.get_columns('products')
print('Product Columns:')
for col in columns:
    null_str = 'NOT NULL' if col.not_null else 'nullable'
    print(f'  {col.name}: {col.type} ({null_str})')

indexes = db.get_indexes('products')
print('Indexes:')
for idx in indexes:
    print(f'  {idx.name} on {idx.columns}')

# Use custom types
db.execute(
    'INSERT INTO products (sku, name, price, cost, added_date) VALUES (?, ?, ?, ?, ?)',
    ('PROD-001', 'Widget', Decimal('29.99'), Decimal('15.00'), date(2024, 1, 1))
)

# Retrieve with automatic conversion
row = db.execute_one('SELECT * FROM products WHERE sku = ?', ('PROD-001',))
print(f'\nRetrieved product:')
print(f'  SKU: {row[1]}')
print(f'  Price: {row[3]} (type: {type(row[3]).__name__})')
print(f'  Cost: {row[4]} (type: {type(row[4]).__name__})')
print(f'  Added: {row[5]} (type: {type(row[5]).__name__})')

db.close()
```

--------------------------------

### Asyncio Database Operations

Source: https://github.com/coleifer/cysqlite/blob/master/_autodocs/README.md

Provides an example of using Cysqlite with asyncio for non-blocking database operations. Requires importing connect from cysqlite.aio.

```python
import asyncio
from cysqlite.aio import connect

async def main():
    db = await connect(':memory:')
    cursor = await db.execute('SELECT 42')
    result = await cursor.scalar()
    await db.close()

asyncio.run(main())
```

--------------------------------

### Read-Heavy Workload Example

Source: https://github.com/coleifer/cysqlite/blob/master/_autodocs/pooling.md

Demonstrates a usage pattern for read-heavy workloads using multiple concurrent reader connections managed by the Pool.

```APIDOC
## Usage Patterns

### Read-Heavy Workload

```python
from cysqlite.utils import Pool
from concurrent.futures import ThreadPoolExecutor
import time

pool = Pool('/data/app.db', readers=16, writer=True)

def process_user(user_id):
    """Read-only operation."""
    with pool.reader() as db:
        return db.execute(
            'SELECT * FROM users WHERE id = ?', (user_id,)
        ).fetchone()

def batch_process():
    """Process many users in parallel."""
    with ThreadPoolExecutor(max_workers=16) as executor:
        user_ids = range(1, 1001)
        results = executor.map(process_user, user_ids)
        for result in results:
            if result:
                print(f'User: {result[1]}')

batch_process()
pool.close()
```
```

--------------------------------

### Execute SQL and Fetch Results

Source: https://github.com/coleifer/cysqlite/blob/master/docs/api.md

Provides examples of executing SQL statements with parameters, including creating tables, bulk inserts with returning values, retrieving single rows, and handling empty result sets.

```python
curs = db.cursor()
curs.execute('create table kv ("id" integer primary key, "key", "value")')
curs.execute('insert into kv (key, value) values (?, ?), (?, ?) '
             'returning id, key', ('k1', 'v1', 'k2', 'v2'))
for (i, k) in curs:
    print(f'inserted {k} with id={i}')
curs.execute('select * from kv where id = :pk', {'pk': 1})
row = curs.fetchone()
print(f'retrieved row 1: {row}')
curs.execute('select * from kv where id = 0')
assert curs.fetchone() is None
assert curs.fetchall() == []
```

--------------------------------

### Manual Transaction with Rollback Example

Source: https://github.com/coleifer/cysqlite/blob/master/_autodocs/transactions.md

Demonstrates explicit transaction control using begin, commit, and rollback. Use this pattern when you need fine-grained control over transaction lifecycles and error handling.

```python
db = connect(':memory:')
db.execute('CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT, balance REAL)')

db.begin('EXCLUSIVE')
try:
    # Transfer funds
    db.execute('UPDATE users SET balance = balance - 100 WHERE id = ?', (1,))
    db.execute('UPDATE users SET balance = balance + 100 WHERE id = ?', (2,))
    db.commit()
except Exception:
    db.rollback()
    raise
```

--------------------------------

### Full Example: Inserting and Reading Converted Data

Source: https://github.com/coleifer/cysqlite/blob/master/docs/api.md

Demonstrates creating a table with custom data types (datetime, json, numeric), inserting data including a timezone-aware datetime, JSON object, and Decimal, and then reading it back, showing automatic conversion.

```python
db.execute('create table vals (ts datetime, js json, dec numeric(10, 2))')

# Create a TZ-aware datetime, a JSON object and a Decimal.
ts = datetime.datetime(2026, 1, 2, 3, 4, 5).astimezone(datetime.timezone.utc)
js = {'key': {'nested': 'value'}, 'arr': ['i0', 1, 2., None]}
dec = Decimal('1.3')

# When we INSERT the JSON, note that we need to dump it to string.
db.execute('insert into vals (ts, js, dec) values (?, ?, ?)',
           (ts, json.dumps(js), dec))

# When reading back the data, it is converted automatically based on
# the declared column types.
row = db.execute_one('select * from vals')
assert row == (ts, js, dec)
```

--------------------------------

### Control File Operations

Source: https://github.com/coleifer/cysqlite/blob/master/docs/api.md

Illustrates using `file_control` to manage file-level operations. This example shows enabling persistent WAL mode and retrieving the data version.

```python
db.file_control(SQLITE_FCNTL_PERSIST_WAL, 1)
version = db.file_control(SQLITE_FCNTL_DATA_VERSION, 0)
print(f'Data version: {version}')
```

--------------------------------

### Execute Scalar Example

Source: https://github.com/coleifer/cysqlite/blob/master/_autodocs/asyncio.md

Execute a query and return the first column of the first row. Useful for retrieving single values like counts.

```python
count = await db.execute_scalar('SELECT COUNT(*) FROM users')
print(f'Total users: {count}')
```

--------------------------------

### Register Custom Data Adapters

Source: https://github.com/coleifer/cysqlite/blob/master/docs/api.md

Shows how to register custom adapters for Python data types to be automatically serialized when used as bound parameters in SQL queries. Includes examples for dict, datetime, and Decimal types.

```python
db = cysqlite.connect(':memory:')

# Automatically serialize dict as JSON.
db.register_adapter(dict, json.dumps)

# Or use the `adapter()` decorator.
@db.adapter(datetime.datetime)
def adapt_datetime(value):
    # Converts datetimes to a UNIX timestamp.
    return value.timestamp()

# Store Decimal as TEXT to avoid floating-point precision errors.
@db.adapter(Decimal)
def adapt_decimal(value):
    return str(value)

vals = [
    {'key': 'value'},
    datetime.datetime(2026, 1, 2, 3, 4, 5),
    Decimal('1.3'),
]
for val in vals:
    res = db.execute_scalar('select ?', (val,))
    print(f'{res} ({type(res)})')

# {"key": "value"} <class 'str'>
# 1767344645.0 <class 'float'>
# 1.3 <class 'str'>
```

--------------------------------

### Example Pool Usage for Safe Transactions

Source: https://github.com/coleifer/cysqlite/blob/master/docs/aio.md

Demonstrates how to use the Pool to serialize access to the writer connection, ensuring transaction safety when multiple tasks might need to write to the database.

```python
pool = Pool('app.db')

async def task_a(pool):
    async with pool.writer() as db:
        async with db.atomic() as tx:
            await db.execute('insert into ...')
            await asyncio.sleep(0)
```

```python
async def task_a(pool):
    async with pool.writer() as db:
        async with db.atomic() as tx:
            await db.execute('insert into ...')
            await tx.rollback()
```

--------------------------------

### Transaction Context Manager Example

Source: https://github.com/coleifer/cysqlite/blob/master/_autodocs/transactions.md

Uses the `transaction()` context manager to automatically commit on success or roll back on exception. This simplifies transaction handling for a block of operations.

```python
with db.transaction():
    db.execute('INSERT INTO users (name) VALUES (?)', ('Alice',))
    db.execute('INSERT INTO users (name) VALUES (?)', ('Bob',))
```

--------------------------------

### Enable Autocommit Mode with autocommit=True

Source: https://github.com/coleifer/cysqlite/blob/master/docs/sqlite-notes.md

Using `autocommit=True` provides a modern way to disable driver-managed transactions in Python 3.12+. This example shows connecting to an in-memory database and manually starting and committing a transaction.

```python
# Let's do it the way modern Python wants us to.
>>> db = sqlite3.connect(':memory:', autocommit=True)

# Isolation level returns an empty string. Empty. It means whatever you
# want it to mean.
>>> db.isolation_level
''

>>> db.in_transaction
False

# So far so good.
>>> db.execute('begin')
>>> db.in_transaction
True
```

--------------------------------

### Savepoint Context Manager Example

Source: https://github.com/coleifer/cysqlite/blob/master/_autodocs/transactions.md

Utilizes the `savepoint()` context manager for partial rollbacks. Operations within the savepoint are rolled back if an exception occurs, while operations outside remain.

```python
db.begin()
try:
    db.execute('INSERT INTO main_data (v) VALUES (?)', (1,))
    
    try:
        with db.savepoint('sp1'):
            db.execute('INSERT INTO temp_data (v) VALUES (?)', (2,))
            raise ValueError('Bad temp data')
    except ValueError:
        pass  # Temp insert rolled back, main_data still there
    
    db.execute('INSERT INTO main_data (v) VALUES (?)', (3,))
    db.commit()
except Exception:
    db.rollback()
```

--------------------------------

### Initialize cysqlite Pool with Custom Pragmas

Source: https://github.com/coleifer/cysqlite/blob/master/_autodocs/pooling.md

Demonstrates creating a Pool instance with a specified database file, number of readers, and custom pragmas for connection configuration.

```python
from cysqlite.utils import Pool

# Create pool with custom pragmas
pool = Pool('/data/app.db',
            readers=8,
            writer=True,
            pragmas={'journal_mode': 'wal',
                     'cache_size': -32000})
```

--------------------------------

### MemStore Table Function (Writable)

Source: https://github.com/coleifer/cysqlite/blob/master/docs/api.md

An example of a writable table function that implements an in-memory key/value store. It demonstrates how to support `INSERT`, `UPDATE`, and `DELETE` operations by implementing the respective methods.

```APIDOC
## MemStore Table Function (Writable)

### Description
This example defines a `MemStore` table function that acts as an in-memory key/value store. It supports `INSERT`, `UPDATE`, and `DELETE` operations, making it a fully functional virtual table.

### Method
`MemStore.register(db)`

### Parameters
None for registration.

### Usage Example
```python
from cysqlite import TableFunction

class MemStore(TableFunction):
    name = 'memstore'
    columns = [('id', 'INTEGER'), ('key', 'TEXT'), ('value', 'TEXT')]
    params = []
    with_rowid = True

    _data = {}
    _next_id = 1

    def initialize(self, **filters):
        pass

    def iterate(self, idx):
        keys = sorted(self._data)
        if idx >= len(keys):
            raise StopIteration
        rowid = keys[idx]
        row = self._data[rowid]
        return (rowid, (row['id'], row['key'], row['value']))

    def insert(self, rowid, values):
        if rowid is None:
            rowid = self._next_id
            MemStore._next_id += 1
        else:
            rowid = int(rowid)
            if rowid >= MemStore._next_id:
                MemStore._next_id = rowid + 1
        if len(values) < 3:
            raise ValueError('Expected 3 values, got %s' % len(values))
        u_rowid, key, value = values
        self._data[rowid] = {
            'id': int(u_rowid) if u_rowid is not None else rowid,
            'key': str(key) if key is not None else key,
            'value': str(value) if value is not None else value,
        }
        return rowid

    def update(self, old_rowid, new_rowid, values):
        old_rowid = int(old_rowid)
        new_rowid = int(new_rowid)
        if old_rowid not in self._data:
            raise ValueError('Row %s not found' % old_rowid)
        if len(values) < 3:
            raise ValueError('Expected 3 values, got %s' % len(values))
        uid, key, value = values
        if uid:
            new_rowid = uid
        if old_rowid != new_rowid:
            self._data[new_rowid] = self._data.pop(old_rowid)
            rowid = new_rowid
        else:
            rowid = old_rowid
        if uid:
            self._data[rowid]['id'] = int(uid)
        if key is not None:
            self._data[rowid]['key'] = key
        if value is not None:
            self._data[rowid]['value'] = value

    def delete(self, rowid):
        rowid = int(rowid)
        if rowid not in self._data:
            raise ValueError('Row %s not found' % rowid)
        del self._data[rowid]

db = connect(':memory:')
MemStore.register(db)
db.execute('insert into memstore (id, key, value) values (?, ?, ?)', (1, 'k1', 'v1'))
db.execute('insert into memstore (key, value) values (?, ?), (?, ?)', ('k2', 'v2', 'k3', 'v3'))
for row in db.execute('select * from memstore where value != ?', ('v2',)):
    print(row)
db.execute('update memstore set value = ? where key = ?', ('v2y', 'k2'))
db.execute('update memstore set value = value || ?', ('z',))
for row in db.execute('select * from memstore'):
    print(row)
```

### Response
- The `insert` method returns the `rowid` of the inserted row.
- The `update` method modifies an existing row.
- The `delete` method removes a row.
```

--------------------------------

### Get Column Value with Default

Source: https://github.com/coleifer/cysqlite/blob/master/_autodocs/row.md

Use the `get()` method to retrieve a column's value, providing a default value if the column does not exist.

```python
phone = row.get('phone', 'N/A')
```

--------------------------------

### Basic Query Execution

Source: https://github.com/coleifer/cysqlite/blob/master/_autodocs/README.md

Demonstrates how to establish a connection, create a table, insert data, and execute a select query. Use this for fundamental database operations.

```python
from cysqlite import connect

db = connect(':memory:')
db.execute('CREATE TABLE users (id INT, name TEXT)')
db.execute('INSERT INTO users VALUES (?, ?)', (1, 'Alice'))

row = db.execute_one('SELECT * FROM users WHERE id = ?', (1,))
print(row)  # (1, 'Alice')

db.close()
```

--------------------------------

### Read-Only Table Function Example

Source: https://github.com/coleifer/cysqlite/blob/master/docs/api.md

Implement a read-only table function to generate a series of numbers. This example demonstrates the `initialize` and `iterate` methods for data generation.

```python
from cysqlite import TableFunction

class Series(TableFunction):
    name = 'series'  # Specify name - if empty class name is used.

    # Name of columns in each row of generated data.
    columns = ['value']

    # Name of parameters the function may be called with.
    params = ['start', 'stop', 'step']

    def initialize(self, start=0, stop=None, step=1):
        """
        Table-functions declare an initialize() method, which is
        called with whatever arguments the user has called the
        function with.
        """
        self.start = self.current = start
        self.stop = stop or float('Inf')
        self.step = step

    def iterate(self, idx):
        """
        Iterate is called repeatedly by the SQLite database engine
        until the required number of rows has been read **or** the
        function raises a `StopIteration` signalling no more rows
        are available.
        """
        if self.current > self.stop:
            raise StopIteration

        ret, self.current = self.current, self.current + self.step
        return (ret,)

# Register the table-function with our database.
db = connect(':memory:')
Series.register(db)

# Usage:
cursor = db.execute('select * from series(?, ?, ?)', (0, 5, 2))
for row in cursor:
    print(row)
# (0,)
# (2,)
# (4,)
```

--------------------------------

### Creating and Querying Database Views

Source: https://github.com/coleifer/cysqlite/blob/master/_autodocs/types.md

Demonstrates how to create a view in the database and then retrieve a list of all defined views.

```python
db.execute('CREATE TABLE orders (id INT, customer_id INT, total REAL)')
db.execute('''
    CREATE VIEW high_value_orders AS
    SELECT id, customer_id, total FROM orders WHERE total > 100
''')

views = db.get_views()
for v in views:
    print(f'{v.name}:
{v.sql}')
```

--------------------------------

### Get Last Inserted Row ID in cysqlite

Source: https://github.com/coleifer/cysqlite/blob/master/_autodocs/connection.md

Retrieve the row ID of the most recently inserted row using `last_insert_rowid()`. This is useful after INSERT operations to get the primary key of the new record.

```python
db.execute('INSERT INTO users (name) VALUES (?)', ('Alice',))
user_id = db.last_insert_rowid()
```

--------------------------------

### Mutable Table Function Example (In-Memory Store)

Source: https://github.com/coleifer/cysqlite/blob/master/docs/api.md

Implement a mutable table function to create an in-memory key/value store. This example shows how to handle INSERT, UPDATE, and DELETE operations, and uses `with_rowid=True`.

```python
class MemStore(TableFunction):
    """
    In-memory key/value store exposed as a virtual-table.
    """
    name = 'memstore'
    columns = [
        ('id', 'INTEGER'),
        ('key', 'TEXT'),
        ('value', 'TEXT')]
    params = []
    with_rowid = True  # iterate() must return (rowid, (row tuple)).

    _data = {}
    _next_id = 1

    def initialize(self, **filters):
        pass

    def iterate(self, idx):
        keys = sorted(self._data)
        if idx >= len(keys):
            raise StopIteration

        rowid = keys[idx]
        row = self._data[rowid]

        # Return a 2-tuple of (rowid, row-data) to specify rowid explicitly.
        return (rowid, (row['id'], row['key'], row['value']))

    def insert(self, rowid, values):
        # rowid might be None, so we auto-generate
        if rowid is None:
            rowid = self._next_id
            MemStore._next_id += 1
        else:
            rowid = int(rowid)
            if rowid >= MemStore._next_id:
                MemStore._next_id = rowid + 1

        if len(values) < 3:
            raise ValueError('Expected 3 values, got %s' % len(values))

        u_rowid, key, value = values
        self._data[rowid] = {
            'id': int(u_rowid) if u_rowid is not None else rowid,
            'key': str(key) if key is not None else key,
            'value': str(value) if value is not None else value,
        }

        return rowid

    def update(self, old_rowid, new_rowid, values):
        old_rowid = int(old_rowid)
        new_rowid = int(new_rowid)

        if old_rowid not in self._data:
            raise ValueError('Row %s not found' % old_rowid)

        if len(values) < 3:
            raise ValueError('Expected 3 values, got %s' % len(values))

        uid, key, value = values
        if uid:
            new_rowid = uid

        if old_rowid != new_rowid:
            # User updated rowid, move data.
            self._data[new_rowid] = self._data.pop(old_rowid)
            rowid = new_rowid
        else:
            rowid = old_rowid

        if uid:
            self._data[rowid]['id'] = int(uid)
        if key is not None:
            self._data[rowid]['key'] = key
        if value is not None:
            self._data[rowid]['value'] = value

    def delete(self, rowid):
        rowid = int(rowid)
        if rowid not in self._data:
            raise ValueError('Row %s not found' % rowid)
        del self._data[rowid]

# Register the table-function with our database.
db = connect(':memory:')
MemStore.register(db)

# Usage:
db.execute('insert into memstore (id, key, value) values (?, ?, ?)',
           (1, 'k1', 'v1'))
db.execute('insert into memstore (key, value) values (?, ?), (?, ?)',
           ('k2', 'v2', 'k3', 'v3'))

assert db.last_insert_rowid() == 3

for row in db.execute('select * from memstore where value != ?', ('v2',)):
    print(row)

# (1, 'k1', 'v1')
# (3, 'k3', 'v3')

db.execute('update memstore set value = ? where key = ?', ('v2y', 'k2'))
db.execute('update memstore set value = value || ?', ('z',))

for row in db.execute('select * from memstore'):
    print(row)

# (1, 'k1', 'v1z')

```

--------------------------------

### Connection Constructor Parameters

Source: https://github.com/coleifer/cysqlite/blob/master/_autodocs/connection.md

Illustrates various ways to instantiate a Connection object, including in-memory databases, persistent databases with custom settings, and URI-based connections.

```python
from cysqlite import connect, Row

# In-memory database
db = connect(':memory:')

# Persistent database with row factory and pragmas
db = connect('/data/app.db',
             row_factory=Row,
             timeout=10.0,
             pragmas={'journal_mode': 'wal', 'cache_size': -64000})

# URI with read-only access
db = connect('file:///data/app.db?mode=ro', uri=True)
```

--------------------------------

### executescript

Source: https://github.com/coleifer/cysqlite/blob/master/docs/api.md

Executes one or more SQL statements in a script. This is useful for setup or complex transactions.

```APIDOC
## executescript(sql)

### Description
Execute the SQL statement(s) in *sql*.

### Parameters
#### Path Parameters
- **sql** (str) - Required - One or more SQL statements to execute in a script.

### Returns
self

### Return type
[`Cursor`](#Cursor)

### Example
```python
curs = db.cursor()
curs.executescript("""
    begin;
    create table users (
       id integer not null primary key,
       name text not null,
       email text not null);
    create index users_email ON users (email);

    create table tweets (
       id integer not null primary key,
       content text not null,
       user_id integer not null references users (id),
       timestamp integer not null);

    commit;
")
```
```

--------------------------------

### Runtime Limits

Source: https://github.com/coleifer/cysqlite/blob/master/docs/api.md

Allows setting and getting values for various run-time limits defined by SQLite.

```APIDOC
## Runtime Limits

### Set or Get Runtime Limit

- `setlimit(category, limit)`: Set the value of a run-time limit.
- `getlimit(category)`: Get the value of a run-time limit.

Limits are defined by [Limit Constants](#limit-constants).

**Example:**
```python
max_sql = db.getlimit(SQLITE_LIMIT_SQL_LENGTH)
db.setlimit(SQLITE_LIMIT_ATTACHED, 3)
```
```

--------------------------------

### Get Number of Columns

Source: https://github.com/coleifer/cysqlite/blob/master/_autodocs/row.md

Use the `len()` function to determine the total number of columns in the row.

```python
num_cols = len(row)
```

--------------------------------

### Interrupting a Long Query

Source: https://github.com/coleifer/cysqlite/blob/master/docs/api.md

Example of how to interrupt a long-running database query using signal handling and db.interrupt().

```python
import signal

def on_ctrl_c(signum, frame):
    db.interrupt()

signal.signal(signal.SIGINT, on_ctrl_c)

try:
    db.execute('select * from big_a, big_b, big_c').fetchall()
except OperationalError:
    print('Query was interrupted.')
```

--------------------------------

### Establish cysqlite Connection

Source: https://github.com/coleifer/cysqlite/blob/master/docs/api.md

Demonstrates various ways to open a cysqlite database connection, including simple connections, in-memory databases, read-only access, and using custom row factories.

```python
conn = cysqlite.Connection('app.db')
```

```python
conn = cysqlite.Connection(':memory:')
```

```python
conn = cysqlite.Connection('data.db',
                           flags=cysqlite.SQLITE_OPEN_READONLY)
```

```python
conn = cysqlite.Connection('app.db', row_factory=cysqlite.Row)
```

--------------------------------

### connect()

Source: https://github.com/coleifer/cysqlite/blob/master/_autodocs/asyncio.md

Create an async database connection (must be called within an event loop).

```APIDOC
## connect()

### Description
Create an async database connection (must be called within an event loop).

### Method
`async def connect(database: str, **kwargs) -> AsyncConnection`

### Parameters
#### Path Parameters
- **database** (str) - Required - Database filename or ":memory:".
- **kwargs** (any) - Optional - Additional arguments passed to `cysqlite.connect()`.

### Response
#### Success Response
- **AsyncConnection** - An `AsyncConnection` instance.

### Example
```python
import asyncio
from cysqlite.aio import connect

async def main():
    db = await connect(':memory:')
    await db.close()

asyncio.run(main())
```
```

--------------------------------

### Get List of Row Values

Source: https://github.com/coleifer/cysqlite/blob/master/_autodocs/row.md

The `values()` method returns a list containing all the values in the row, in their original order.

```python
row.values()
```

--------------------------------

### Get Database Status

Source: https://github.com/coleifer/cysqlite/blob/master/docs/api.md

Retrieves and prints the current and highwater values for the cache usage status flag of the database connection.

```python
db = connect('app.db')

# Execute a query.
db.execute('select * from register').fetchall()

# Get the amount of memory used by cached data (pages, schema, etc).
current, highwater = db.status(SQLITE_DBSTATUS_CACHE_USED)
print(f'Cache: {current} bytes (peak: {highwater})')
```

--------------------------------

### Get List of Column Names

Source: https://github.com/coleifer/cysqlite/blob/master/_autodocs/row.md

The `keys()` method returns a list of all column names present in the row, in their original order.

```python
cols = row.keys()
print(f'Columns: {cols}')
```

--------------------------------

### Initialize Async Connection Pool

Source: https://github.com/coleifer/cysqlite/blob/master/_autodocs/asyncio.md

Create an asyncio connection pool with separate reader and writer connections. Specify the database file, number of readers, and whether to create a writer connection.

```python
from cysqlite.aio import Pool

pool = await Pool(':memory:', readers=4, writer=True)
```

--------------------------------

### Define Table Function Columns with Data Types

Source: https://github.com/coleifer/cysqlite/blob/master/docs/api.md

Example of defining columns for a TableFunction with specified names and data types.

```python
columns = [('key', 'text'), ('data', 'blob')]
```

--------------------------------

### Import and Use cysqlite Module Exports

Source: https://github.com/coleifer/cysqlite/blob/master/_autodocs/module-exports.md

Demonstrates importing and utilizing various symbols from the cysqlite library, including connection objects, cursors, exceptions, utility functions, and constants. Shows how to connect to an in-memory database, check compilation options, list VFS, and handle integrity errors.

```python
from cysqlite import (
    connect,
    Connection,
    Cursor,
    Row,
    Blob,
    # Exceptions
    IntegrityError,
    OperationalError,
    DatabaseLockedError,
    # Functions
    compile_option,
    vfs_list,
    # Constants
    SQLITE_OK,
    SQLITE_CONSTRAINT,
    SQLITE_OPEN_READONLY,
)

from cysqlite.aio import connect as async_connect, Pool as AsyncPool
from cysqlite.utils import Pool, slow_query_log
from cysqlite.metadata import Column, Index, ForeignKey

# Use exported symbols
db = connect(':memory:')
print(f'SQLite version: {db.pragma("database_list")}')

# Check if SQLite was compiled with specific option
print(f'FTS5 support: {compile_option("enable_fts5")}')

# List available VFS implementations
print(f'Available VFS: {vfs_list()}')

try:
    db.execute('INSERT INTO users VALUES (1, "Alice")')
except IntegrityError as e:
    print(f'Constraint violated: {e}')

db.close()
```

--------------------------------

### Get Column Names

Source: https://github.com/coleifer/cysqlite/blob/master/_autodocs/cursor.md

Fetch a list of column names from the current result set. Useful for understanding the structure of returned data.

```python
cursor.execute('SELECT id, name, email FROM users')
cols = cursor.columns()
print(f'Columns: {cols}')  # ['id', 'name', 'email']
```

--------------------------------

### Get Last Inserted Row ID

Source: https://github.com/coleifer/cysqlite/blob/master/_autodocs/cursor.md

Retrieve the ID of the last row inserted into a table. Returns None if no inserts have occurred.

```python
cursor.execute('INSERT INTO users (name) VALUES (?)', ('Alice',))
user_id = cursor.lastrowid
print(f'New user ID: {user_id}')
```

--------------------------------

### BLOB Handling with zeroblob and blob_open

Source: https://github.com/coleifer/cysqlite/blob/master/docs/api.md

Demonstrates creating a BLOB of a fixed size using zeroblob and then accessing and modifying it using the blob_open handle. Note that the BLOB size cannot be changed after creation.

```python
db.execute('create table register (
' \
           'id integer primary key, ' \
           'data blob not null)')

# Create a row with a 16-byte empty blob.
db.execute('insert into register (data) values (zeroblob(?))', (16,))
rowid = db.last_insert_rowid()

# Obtain a handle to access the BLOB.
blob = db.blob_open('register', 'data', rowid)

blob.write(b'abcdefgh')
assert blob.tell() == 8

blob.seek(2)
print(blob.read(4))  # b'cdef'

blob.close()  # Release the blob handle.
```

--------------------------------

### begin

Source: https://github.com/coleifer/cysqlite/blob/master/docs/aio.md

Begin a new transaction. Supports different lock types for controlling concurrency.

```APIDOC
## begin

### Description
Begin a transaction.
If a transaction is already active, raises [`OperationalError`](api.md#OperationalError).

### Method
async

### Parameters
#### Parameters
- **lock** (str) – type of SQLite lock to acquire, `DEFERRED` (default), `IMMEDIATE`, or `EXCLUSIVE`.
```

--------------------------------

### Get Current BLOB Position

Source: https://github.com/coleifer/cysqlite/blob/master/_autodocs/blob.md

Retrieves the current read/write position within the BLOB. This is useful for tracking progress or resuming operations.

```python
pos = blob.tell()
print(f'Currently at byte {pos}')
```

--------------------------------

### Creating a Table with a Foreign Key Constraint

Source: https://github.com/coleifer/cysqlite/blob/master/_autodocs/types.md

Shows how to create a table with a foreign key constraint, specifying the action to take on deletion.

```python
db.execute('CREATE TABLE customers (id INT PRIMARY KEY, name TEXT)')
db.execute('''
    CREATE TABLE orders (
        id INT PRIMARY KEY,
        customer_id INT,
        FOREIGN KEY(customer_id) REFERENCES customers(id) ON DELETE CASCADE
    )
''')

fks = db.get_foreign_keys('orders')
for fk in fks:
    print(f'{fk.column} → {fk.referred_table}({fk.referred_column})')
    print(f'  ON DELETE: {fk.on_delete}')
```

--------------------------------

### Get Column Metadata

Source: https://github.com/coleifer/cysqlite/blob/master/docs/api.md

Fetch detailed metadata for a specific column in a table using table_column_metadata(). Returns a ColumnMetadata object with comprehensive information.

```python
db.table_column_metadata('entry', 'title')
ColumnMetadata(
   table='entry',
   column='title',
   datatype='TEXT',
   collation='BINARY',
   not_null=True,
   primary_key=False,
   auto_increment=False)
```

--------------------------------

### Basic Connection

Source: https://github.com/coleifer/cysqlite/blob/master/docs/migration.md

Connect to a SQLite database file. This is the most basic way to establish a connection.

```python
db = sqlite3.connect('app.db')
```

```python
db = cysqlite.connect('app.db')
```

--------------------------------

### Get Foreign Keys

Source: https://github.com/coleifer/cysqlite/blob/master/docs/api.md

List the foreign keys defined for a table using get_foreign_keys(). The output is a list of ForeignKey tuples detailing the relationship.

```python
print(db.get_foreign_keys('entrytag'))
[ForeignKey(
     column='entry_id',
     dest_table='entry',
     dest_column='id',
     table='entrytag'),
 ...]
```

--------------------------------

### Optimal Settings for Development and Testing

Source: https://github.com/coleifer/cysqlite/blob/master/_autodocs/pragmas-and-limits.md

Use an in-memory database for development and testing to ensure fast, isolated environments. This configuration is ideal for rapid iteration and testing without persistent storage.

```python
db = connect(':memory:',
             pragmas={
                 'foreign_keys': 1,
                 'temp_store': 'memory',
            })

```

--------------------------------

### Get Database Views

Source: https://github.com/coleifer/cysqlite/blob/master/docs/api.md

Retrieve a list of all views in the database using get_views(). The output includes View objects with name and SQL definition.

```python
print(db.get_views())
[View(name='entries_public',
      sql='CREATE VIEW entries_public AS SELECT ... '),
 View(...),
 ...]
```

--------------------------------

### Configuring Performance PRAGMAs

Source: https://github.com/coleifer/cysqlite/blob/master/_autodocs/pragmas-and-limits.md

Optimize database performance by setting cache size, memory-mapped I/O size, and page size when connecting.

```python
db = connect('/data/app.db', pragmas={
    'cache_size': -64000,
    'mmap_size': 256 * 1024 * 1024,
})
```

--------------------------------

### Get Foreign Key Constraints

Source: https://github.com/coleifer/cysqlite/blob/master/_autodocs/connection.md

Retrieves a list of foreign key constraints defined for a table. Important for understanding referential integrity between tables.

```python
def get_foreign_keys(table: str, database=None) -> list[ForeignKey]
```

--------------------------------

### Using cysqlite.Row for Flexible Data Access

Source: https://github.com/coleifer/cysqlite/blob/master/docs/api.md

Demonstrates how to set cysqlite.Row as the row factory and access data by attribute, column name, or index. Also shows dict-like interface methods and membership testing.

```python
db = cysqlite.connect(':memory:')
db.row_factory = cysysqlite.Row

db.execute('create table users (id integer primary key, ' \
           'username text, active integer)')
db.execute('insert into users values (?, ?, ?)', (1, 'charles', 1))

row = db.execute('select * from users').fetchone()

# repr shows all columns and values.
print(row)
# <Row(id=1, username='charles', active=1)>

# Access by attribute name, column name, or index.
print(row.username)    # 'charles'
print(row['active'])   # 1
print(row[0])          # 1

# Iterate over values.
print(list(row))       # [1, 'charles', 1]

# Dict-like interface.
print(row.keys())
# ['id', 'username', 'active']
print(row.as_dict())   # {'id': 1, 'username': 'charles', 'active': 1}

# Test membership and safe access.
print('username' in row)        # True
print(row.get('email', 'n/a'))  # 'n/a'

# Rows are usable in sets and as dict keys.
seen = set()
for row in db.execute('select * from users'):
    seen.add(row)
```

--------------------------------

### Configuring WAL Mode and Checkpointing

Source: https://github.com/coleifer/cysqlite/blob/master/_autodocs/pragmas-and-limits.md

Enable Write-Ahead Logging (WAL) mode and configure auto-checkpointing frequency for improved concurrency and performance.

```python
db = connect('/data/app.db', journal_mode='wal')
```

```python
db.pragma('wal_autocheckpoint', 5000)  # Checkpoint every 5000 pages
```

--------------------------------

### Manual Transaction Management

Source: https://github.com/coleifer/cysqlite/blob/master/docs/api.md

Demonstrates manual transaction control using begin, commit, and checking autocommit status.

```python
db = connect(':memory:')

assert db.autocommit()  # Autocommit mode by default.

db.begin()  # Now we are in a transaction.
assert not db.autocommit()
db.commit()

assert db.autocommit()  # Back in autocommit mode.
```

--------------------------------

### Manual Transaction Rollback Example

Source: https://github.com/coleifer/cysqlite/blob/master/_autodocs/transactions.md

Rolls back a transaction if a condition is not met after an initial delete operation. This pattern is useful for conditional commits.

```python
db.begin()
try:
    db.execute('DELETE FROM users WHERE id = ?', (42,))
    # Oops, verify before committing
    if db.execute_scalar('SELECT COUNT(*) FROM users') < 10:
        raise ValueError('Too few users remain')
    db.commit()
except Exception:
    db.rollback()
```

--------------------------------

### Manual Transaction Commit Example

Source: https://github.com/coleifer/cysqlite/blob/master/_autodocs/transactions.md

Commits a transaction after executing an insert statement. Ensure all operations within the transaction are valid before calling commit.

```python
db.begin()
db.execute('INSERT INTO logs (msg) VALUES (?)', ('event',))
db.commit()
```

--------------------------------

### connect()

Source: https://github.com/coleifer/cysqlite/blob/master/docs/api.md

Opens a connection to the database. Restores previously registered items but not hooks.

```APIDOC
## connect()

### Description
Open connection to the database.

### Returns
`True` if database was previously closed and is now open. If database was already open returns `False`.

### Return type
bool

### Raises
[`OperationalError`](#OperationalError) if opening database fails.

### Notes
All previously-registered user-defined functions, aggregates, window functions, collations and table functions will be restored, along with any permanent pragmas. Hooks (commit, rollback, update, authorizer, trace and progress) are not restored and must be re-registered.
```

--------------------------------

### Build cysqlite with SQLCipher support

Source: https://github.com/coleifer/cysqlite/blob/master/README.md

Build cysqlite from source, embedding SQLCipher for encryption support. This involves fetching the SQLCipher amalgamation script and then installing.

```shell
# Obtain checkout of cysqlite.
git clone https://github.com/coleifer/cysqlite

# Automatically download latest source amalgamation.
cd cysqlite/
./scripts/fetch_sqlcipher  # Will add sqlite3.c and sqlite3.h in checkout.

# Build self-contained cysqlite with SQLCipher embedded.
SQLCIPHER=1 pip install .
```

--------------------------------

### Handling Large Binary Data with cysqlite Blob

Source: https://github.com/coleifer/cysqlite/blob/master/docs/api.md

Shows how to allocate space for a blob using zeroblob and then incrementally write data to it using the Blob class, which implements io.RawIOBase.

```python
db = connect(':memory:')
db.execute('create table raw_data (id integer primary key, data blob)')

# Allocate 100MB of space for writing a large file incrementally:
db.execute('insert into raw_data (data) values (zeroblob(?))',
           (1024 * 1024 * 100,))
rowid = db.last_insert_rowid()

# Now we can open the row for incremental I/O:
blob = Blob(db, 'raw_data', 'data', rowid)

# Read from the file and write to the blob in chunks of 4096 bytes.
while True:
    data = file_handle.read(4096)
    if not data:
        break
    blob.write(data)

bytes_written = blob.tell()
blob.close()
```

--------------------------------

### Set and Get Runtime Limits

Source: https://github.com/coleifer/cysqlite/blob/master/docs/api.md

Shows how to retrieve and modify various runtime limits of the SQLite database, such as SQL length or the number of attached databases.

```python
max_sql = db.getlimit(SQLITE_LIMIT_SQL_LENGTH)
print(f'Max SQL length: {max_sql}')
old = db.setlimit(SQLITE_LIMIT_ATTACHED, 3)
print(f'Previous limit was: {old}')
```

--------------------------------

### Get Primary Keys

Source: https://github.com/coleifer/cysqlite/blob/master/docs/api.md

Identify the primary key columns of a table using get_primary_keys(). Returns a list of column names that form the primary key.

```python
print(db.get_primary_keys('entry'))
['id']
```

--------------------------------

### Context Manager Usage

Source: https://github.com/coleifer/cysqlite/blob/master/_autodocs/blob.md

Demonstrates using the BLOB object as a context manager for automatic cleanup.

```APIDOC
## Context Manager

### Description
Use the BLOB object as a context manager to ensure automatic cleanup (closing the BLOB) upon exiting the block.

### Example
```python
with db.blob_open('media', 'image', rowid=1) as blob:
    data = blob.read()
# BLOB is automatically closed here
```
```

--------------------------------

### Get SQLite Memory Usage Status

Source: https://github.com/coleifer/cysqlite/blob/master/docs/api.md

Retrieves the current and peak memory usage of SQLite across all connections. Requires importing status flags.

```python
from cysqlite import status, SQLITE_STATUS_MEMORY_USED

# How much memory has SQLite allocated across all connections?
current, highwater = status(SQLITE_STATUS_MEMORY_USED)
print(f'SQLite is using {current} bytes (peak: {highwater})')
```