SQLAlchemy Core

https://docs.sqlalchemy.org/en/20/core/
Admin
SQLAlchemy Core is a SQL rendering engine and schema-centric toolkit providing database abstraction,...
Tokens:1,381,887
Snippets:8,270
Trust Score:9.7
Update:2 days ago
Show doc for...
Tokens:
Raw
### SQL generated by column() example

Source: https://docs.sqlalchemy.org/en/20/core/sqlelement.html

The SQL output generated by the previous column() example.

```sql
SELECT id, name FROM user
```

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

### before_create event listener example

Source: https://docs.sqlalchemy.org/en/20/core/events.html

Example listener for the 'before_create' event using @event.listens_for. The listener receives target, connection, and **kw arguments.

```python
from sqlalchemy import event


@event.listens_for(SomeSchemaClassOrObject, 'before_create')
def receive_before_create(target, connection, **kw):
    "listen for the 'before_create' event"

    # ... (event handling logic) ...
```

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

### Generated SQL for asc() example

Source: https://docs.sqlalchemy.org/en/20/core/sqlelement.html

Shows the SQL produced by the asc() example: SELECT id, name FROM user ORDER BY name ASC.

```sql
SELECT id, name FROM user ORDER BY name ASC
```

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

### Select.join() example

Source: https://docs.sqlalchemy.org/en/20/core/selectable.html

Creates a SQL JOIN using Select.join() with an onclause. The example shows joining user_table with address_table on matching id columns.

```python
stmt = select(user_table).join(
    address_table, user_table.c.id == address_table.c.user_id
)
```

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

### over() range_ examples

Source: https://docs.sqlalchemy.org/en/20/core/sqlelement.html

Examples of using range_, rows, and groups parameters with over() to specify window frame boundaries.

```python
func.row_number().over(order_by="x", range_=(-5, 10))
```

```python
func.row_number().over(order_by="x", rows=(None, 0))
```

```python
func.row_number().over(order_by="x", range_=(-2, None))
```

```python
func.row_number().over(order_by="x", range_=(1, 3))
```

```python
func.row_number().over(order_by="x", groups=(1, 3))
```

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

### Table.insert() example

Source: https://docs.sqlalchemy.org/en/20/core/metadata.html

Shows how to generate an Insert construct using the Table.insert() method. The example calls .values() to set column values.

```python
table.insert().values(name="foo")
```

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

### Resulting SQL from Table.join() example

Source: https://docs.sqlalchemy.org/en/20/core/metadata.html

The SQL emitted by the join example above, showing a JOIN between user and address tables.

```sql
SELECT user.id, user.name FROM user
JOIN address ON user.id = address.user_id
```

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

### Rendered SQL for tablesample() example

Source: https://docs.sqlalchemy.org/en/20/core/selectable.html

Shows the SQL rendered from the previous Python example, including the TABLESAMPLE bernoulli clause and the REPEATABLE sub-clause with random().

```sql
SELECT alias.people_id FROM
people AS alias TABLESAMPLE bernoulli(:bernoulli_1)
REPEATABLE (random())
```

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

### Generated SQL for between() example

Source: https://docs.sqlalchemy.org/en/20/core/sqlelement.html

Shows the SQL resembling the between() example: SELECT id, name FROM user WHERE id BETWEEN :id_1 AND :id_2.

```sql
SELECT id, name FROM user WHERE id BETWEEN :id_1 AND :id_2
```

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

### before_drop event listener example

Source: https://docs.sqlalchemy.org/en/20/core/events.html

Example listener for the 'before_drop' event using @event.listens_for. The listener receives target, connection, and **kw arguments.

```python
from sqlalchemy import event


@event.listens_for(SomeSchemaClassOrObject, 'before_drop')
def receive_before_drop(target, connection, **kw):
    "listen for the 'before_drop' event"

    # ... (event handling logic) ...
```

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

### istartswith usage example

Source: https://docs.sqlalchemy.org/en/20/core/metadata.html

Example of using istartswith in a select statement to filter rows where a column starts with a given string, case-insensitively.

```python
stmt = select(sometable).where(sometable.c.column.istartswith("foobar"))
```

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

### setup.cfg entry point for a new dialect

Source: https://docs.sqlalchemy.org/en/20/core/connections.html

Defines a setuptools entry point that maps the 'foodialect://' URL to the FooDialect class in the foodialect.dialect module.

```ini
[options.entry_points]
sqlalchemy.dialects =
    foodialect = foodialect.dialect:FooDialect

```

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

### Using startswith in a select statement

Source: https://docs.sqlalchemy.org/en/20/core/metadata.html

Example of using startswith in a SQLAlchemy select() to filter rows where a column starts with a given string.

```python
stmt = select(sometable).where(sometable.c.column.startswith("foobar"))
```

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

### Example of dialect provisioning with module import

Source: https://docs.sqlalchemy.org/en/20/core/internals.html

Illustrates a provisioning method that constructs a module name and imports it, with a fallback for ImportError.

```python
@classmethod
def load_provisioning(cls):
    package = ".".join(cls.__module__.split(".")[0:-1])
    try:
        __import__(package + ".provision")
    except ImportError:
        pass

```

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

### Iterate over sorted_tables to print table names

Source: https://docs.sqlalchemy.org/en/20/core/metadata.html

Use metadata_obj.sorted_tables to get tables in order of foreign key dependency. The example prints each table name, showing the dependency order.

```python
>>> for t in metadata_obj.sorted_tables:
...     print(t.name)
user
user_preference
invoice
invoice_item
```

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

### setup.cfg entry point for a database-qualified dialect

Source: https://docs.sqlalchemy.org/en/20/core/connections.html

Defines an entry point for a MySQL dialect, making it accessible via 'mysql+foodialect://'.

```ini
[options.entry_points]
sqlalchemy.dialects
    mysql.foodialect = foodialect.dialect:FooDialect

```

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

### In-process dialect registration with registry.register()

Source: https://docs.sqlalchemy.org/en/20/core/connections.html

Registers a dialect within the current process, bypassing installation. The example registers 'mysql.foodialect' to load MyMySQLDialect from myapp.dialect, responding to create_engine('mysql+foodialect://').

```python
from sqlalchemy.dialects import registry

registry.register("mysql.foodialect", "myapp.dialect", "MyMySQLDialect")

```

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

### Typical load_provisioning implementation

Source: https://docs.sqlalchemy.org/en/20/core/internals.html

Shows a typical implementation of the load_provisioning classmethod for a dialect that includes a provision.py module. It imports the provisioning module to initiate the setup process.

```python
@classmethod
def load_provisioning(cls):
    __import__("mydialect.provision")
```

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

### Configure Sentinel Column with orm_insert_sentinel

Source: https://docs.sqlalchemy.org/en/20/core/connections.html

Defines a declarative base with a declared_attr that adds a sentinel integer column named '_sentinel' to all mapped classes. This column is used by the 'insertmanyvalues' feature to optimize bulk inserts. The example includes a joined table inheritance setup with MyClass and MySubClass, and a single-table inheritance subclass MySingleInhClass.

```python
from sqlalchemy.orm import declared_attr
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column
from sqlalchemy.orm import orm_insert_sentinel


class Base(DeclarativeBase):
    @declared_attr
    def _sentinel(cls) -> Mapped[int]:
        return orm_insert_sentinel()


class MyClass(Base):
    __tablename__ = "my_table"

    id: Mapped[str] = mapped_column(primary_key=True, server_default=FetchedValue())
    data: Mapped[str] = mapped_column(String(50))


class MySubClass(MyClass):
    __tablename__ = "sub_table"

    id: Mapped[str] = mapped_column(ForeignKey("my_table.id"), primary_key=True)


class MySingleInhClass(MyClass):
    pass
```

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

### CreateEnginePlugin.__init__()

Source: https://docs.sqlalchemy.org/en/20/core/connections.html

Construct a new CreateEnginePlugin instance.

```APIDOC
## CreateEnginePlugin.__init__()

### Description
Construct a new CreateEnginePlugin instance.

### Method
__init__

### Endpoint
CreateEnginePlugin.__init__()

### Parameters
None

### Returns
None
```

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

### iendswith usage example

Source: https://docs.sqlalchemy.org/en/20/core/metadata.html

Example of using iendswith in a select statement.

```python
stmt = select(sometable).where(sometable.c.column.iendswith("foobar"))
```

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

### Engine.begin()

Source: https://docs.sqlalchemy.org/en/20/core/connections.html

Returns a context manager that yields a Connection and begins a transaction. The transaction is committed on success and rolled back on exception.

```APIDOC
## Engine.begin()

### Description
Returns a context manager that yields a Connection and begins a transaction. The transaction is committed on success and rolled back on exception.

### Method
begin

### Signature
`Engine.begin()`

### Returns
- **ContextManager[Connection]** - A context manager yielding a Connection.
```

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

### Connection.begin()

Source: https://docs.sqlalchemy.org/en/20/core/connections.html

Begins a transaction prior to autobegin occurring.

```APIDOC
## Connection.begin()

### Description
Begin a transaction prior to autobegin occurring.

### Method
Method

### Signature
`Connection.begin()`

```

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

### like operator example

Source: https://docs.sqlalchemy.org/en/20/core/metadata.html

Example of using the like operator in a select statement.

```python
stmt = select(sometable).where(sometable.c.column.like("%foobar%"))
```

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

### Engine.begin()

Source: https://docs.sqlalchemy.org/en/20/core/connections.html

Returns a context manager that delivers a Connection with a Transaction established. The transaction is committed on success and rolled back on error.

```APIDOC
## Engine.begin()

### Description
Returns a context manager that delivers a Connection with a Transaction established. The transaction is committed on success and rolled back on error.

### Method
begin()

### Signature
`sqlalchemy.engine.Engine.begin() -> Iterator[Connection]`

### Parameters
None

### Returns
Iterator[Connection] - A context manager yielding a Connection with an active Transaction.

### Example
```python
with engine.begin() as conn:
    conn.execute(text("insert into table (x, y, z) values (1, 2, 3)"))
    conn.execute(text("my_special_procedure(5)"))
```

### Notes
Upon successful operation, the Transaction is committed. If an error is raised, the Transaction is rolled back.

### See Also
- Engine.connect() - procure a Connection from an Engine.
- Connection.begin() - start a Transaction for a particular Connection.
```

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

### Example of dialect provisioning method

Source: https://docs.sqlalchemy.org/en/20/core/internals.html

Shows a typical implementation of the provisioning method for a dialect, including a classmethod decorator and a try-except block for importing a module.

```python
@classmethod
def load_provisioning(cls):
    __import__("mydialect.provision")

```

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

### CTE Usage Examples

Source: https://docs.sqlalchemy.org/en/20/core/selectable.html

Examples of using CTEs in SELECT statements, including non-recursive and recursive CTEs.

```APIDOC
## CTE Usage Examples

### Description
Examples of using CTEs in SELECT statements, including non-recursive and recursive CTEs.

### Example 1: Non-recursive CTE
```python
from sqlalchemy import (
    Table,
    Column,
    String,
    Integer,
    MetaData,
    select,
    func,
)

metadata = MetaData()

orders = Table(
    "orders",
    metadata,
    Column("region", String),
    Column("amount", Integer),
    Column("product", String),
    Column("quantity", Integer),
)

regional_sales = (
    select(orders.c.region, func.sum(orders.c.amount).label("total_sales"))
    .group_by(orders.c.region)
    .cte("regional_sales")
)

top_regions = (
    select(regional_sales.c.region)
    .where(
        regional_sales.c.total_sales
        > select(func.sum(regional_sales.c.total_sales) / 10)
    )
    .cte("top_regions")
)

statement = (
    select(
        orders.c.region,
        orders.c.product,
        func.sum(orders.c.quantity).label("product_units"),
        func.sum(orders.c.amount).label("product_sales"),
    )
    .where(orders.c.region.in_(select(top_regions.c.region)))
    .group_by(orders.c.region, orders.c.product)
)

result = conn.execute(statement).fetchall()
```

### Example 2: WITH RECURSIVE
```python
from sqlalchemy import (
    Table,
    Column,
    String,
    Integer,
    MetaData,
    select,
    func,
)

metadata = MetaData()

parts = Table(
    "parts",
    metadata,
    Column("part", String),
    Column("sub_part", String),
    Column("quantity", Integer),
)

included_parts = (
    select(parts.c.sub_part, parts.c.part, parts.c.quantity)
    .where(parts.c.part == "our part")
    .cte(recursive=True)
)

incl_alias = included_parts.alias()
parts_alias = parts.alias()
included_parts = included_parts.union_all(
    select(
        parts_alias.c.sub_part, parts_alias.c.part, parts_alias.c.quantity
    ).where(parts_alias.c.part == incl_alias.c.sub_part)
)

statement = select(
    included_parts.c.sub_part,
    func.sum(included_parts.c.quantity).label("total_quantity"),
).group_by(included_parts.c.sub_part)

result = conn.execute(statement).fetchall()
```
```

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

### Engine.begin()

Source: https://docs.sqlalchemy.org/en/20/core/connections.html

Returns a context manager that begins a transaction and commits or rolls back on exit.

```APIDOC
## Engine.begin()

### Description
Returns a context manager that begins a transaction and commits or rolls back on exit.

### Method
begin

### Endpoint
Engine.begin()

### Parameters
None

### Returns
A context manager.
```

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

### CTE Usage Examples

Source: https://docs.sqlalchemy.org/en/20/core/selectable.html

Examples of using CTEs in SQLAlchemy, including non-recursive, recursive, and upsert scenarios.

```APIDOC
## CTE Usage Examples

### Description
Examples of using CTEs in SQLAlchemy, including non-recursive, recursive, and upsert scenarios.

### Example 1: Non-recursive CTE
```python
from sqlalchemy import (
    Table,
    Column,
    String,
    Integer,
    MetaData,
    select,
    func,
)

metadata = MetaData()

orders = Table(
    "orders",
    metadata,
    Column("region", String),
    Column("amount", Integer),
    Column("product", String),
    Column("quantity", Integer),
)

regional_sales = (
    select(orders.c.region, func.sum(orders.c.amount).label("total_sales"))
    .group_by(orders.c.region)
    .cte("regional_sales")
)

top_regions = (
    select(regional_sales.c.region)
    .where(
        regional_sales.c.total_sales
        > select(func.sum(regional_sales.c.total_sales) / 10)
    )
    .cte("top_regions")
)

statement = (
    select(
        orders.c.region,
        orders.c.product,
        func.sum(orders.c.quantity).label("product_units"),
        func.sum(orders.c.amount).label("product_sales"),
    )
    .where(orders.c.region.in_(select(top_regions.c.region)))
    .group_by(orders.c.region, orders.c.product)
)

result = conn.execute(statement).fetchall()
```

### Example 2: WITH RECURSIVE
```python
from sqlalchemy import (
    Table,
    Column,
    String,
    Integer,
    MetaData,
    select,
    func,
)

metadata = MetaData()

parts = Table(
    "parts",
    metadata,
    Column("part", String),
    Column("sub_part", String),
    Column("quantity", Integer),
)

included_parts = (
    select(parts.c.sub_part, parts.c.part, parts.c.quantity)
    .where(parts.c.part == "our part")
    .cte(recursive=True)
)

incl_alias = included_parts.alias()
parts_alias = parts.alias()
included_parts = included_parts.union_all(
    select(
        parts_alias.c.sub_part, parts_alias.c.part, parts_alias.c.quantity
    ).where(parts_alias.c.part == incl_alias.c.sub_part)
)

statement = select(
    included_parts.c.sub_part,
    func.sum(included_parts.c.quantity).label("total_quantity"),
).group_by(included_parts.c.sub_part)

result = conn.execute(statement).fetchall()
```

### Example 3: Upsert using UPDATE and INSERT with CTEs
(Example not fully provided in source)
```

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

### Explicit transaction with Connection.begin()

Source: https://docs.sqlalchemy.org/en/20/core/connections.html

Shows how to create an engine, connect, begin a transaction, execute an insert, and commit. The Transaction object is not threadsafe.

```python
from sqlalchemy import create_engine

engine = create_engine("postgresql+psycopg2://scott:tiger@localhost/test")
connection = engine.connect()
trans = connection.begin()
connection.execute(text("insert into x (a, b) values (1, 2)"))
trans.commit()
```

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

### Generated SQL for scalar between() example

Source: https://docs.sqlalchemy.org/en/20/core/sqlelement.html

Shows the SQL produced by the scalar between() example: :param_1 BETWEEN :param_2 AND :param_3.

```sql
:param_1 BETWEEN :param_2 AND :param_3
```

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

### ilike usage example

Source: https://docs.sqlalchemy.org/en/20/core/metadata.html

Example of using ilike in a select statement to perform a case-insensitive LIKE search.

```python
stmt = select(sometable).where(sometable.c.column.ilike("%foobar%"))
```

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

### FromClause.join() example

Source: https://docs.sqlalchemy.org/en/20/core/selectable.html

Shows how to use FromClause.join() to create a join between user_table and address_table, then select from the joined result. The emitted SQL is also shown.

```python
from sqlalchemy import join

j = user_table.join(
    address_table, user_table.c.id == address_table.c.user_id
)
stmt = select(user_table).select_from(j)
```

```sql
SELECT user.id, user.name FROM user
JOIN address ON user.id = address.user_id
```

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

### URL.update_query_pairs example

Source: https://docs.sqlalchemy.org/en/20/core/engines.html

Example from the SQLAlchemy Engine Creation API documentation showing how to use URL.update_query_pairs to update the query parameters of a URL. The example creates a URL from a string, updates it with a list of key-value pairs, and prints the resulting URL string.

```python
>>> from sqlalchemy.engine import make_url
>>> url = make_url("postgresql+psycopg2://user:pass@host/dbname")
>>> url = url.update_query_pairs(
...     [
...         ("alt_host", "host1"),
...         ("alt_host", "host2"),
...         ("ssl_cipher", "/path/to/crt"),
...     ]
... )
>>> str(url)
'postgresql+psycopg2://user:pass@host/dbname?alt_host=host1&alt_host=host2&ssl_cipher=%2Fpath%2Fto%2Fcrt'
```

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

### begin event listener example

Source: https://docs.sqlalchemy.org/en/20/core/events.html

Registers a listener for the 'begin' event on an engine. The listener receives the connection object and can contain event handling logic.

```python
from sqlalchemy import event


@event.listens_for(SomeEngine, 'begin')
def receive_begin(conn):
    "listen for the 'begin' event"

    # ... (event handling logic) ...
```

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

### CTE Examples

Source: https://docs.sqlalchemy.org/en/20/core/selectable.html

Examples of using CTEs in SQLAlchemy, including non-recursive, recursive, upsert, and nested CTE scenarios.

```APIDOC
## CTE Examples

### Description
These examples demonstrate how to use CTEs in SQLAlchemy for various scenarios.

### Example 1: Non-recursive CTE

```python
from sqlalchemy import (
    Table,
    Column,
    String,
    Integer,
    MetaData,
    select,
    func,
)

metadata = MetaData()

orders = Table(
    "orders",
    metadata,
    Column("region", String),
    Column("amount", Integer),
    Column("product", String),
    Column("quantity", Integer),
)

regional_sales = (
    select(orders.c.region, func.sum(orders.c.amount).label("total_sales"))
    .group_by(orders.c.region)
    .cte("regional_sales")
)


top_regions = (
    select(regional_sales.c.region)
    .where(
        regional_sales.c.total_sales
        > select(func.sum(regional_sales.c.total_sales) / 10)
    )
    .cte("top_regions")
)

statement = (
    select(
        orders.c.region,
        orders.c.product,
        func.sum(orders.c.quantity).label("product_units"),
        func.sum(orders.c.amount).label("product_sales"),
    )
    .where(orders.c.region.in_(select(top_regions.c.region)))
    .group_by(orders.c.region, orders.c.product)
)

result = conn.execute(statement).fetchall()
```

### Example 2: WITH RECURSIVE

```python
from sqlalchemy import (
    Table,
    Column,
    String,
    Integer,
    MetaData,
    select,
    func,
)

metadata = MetaData()

parts = Table(
    "parts",
    metadata,
    Column("part", String),
    Column("sub_part", String),
    Column("quantity", Integer),
)

included_parts = (
    select(parts.c.sub_part, parts.c.part, parts.c.quantity)
    .where(parts.c.part == "our part")
    .cte(recursive=True)
)


incl_alias = included_parts.alias()
parts_alias = parts.alias()
included_parts = included_parts.union_all(
    select(
        parts_alias.c.sub_part, parts_alias.c.part, parts_alias.c.quantity
    ).where(parts_alias.c.part == incl_alias.c.sub_part)
)

statement = select(
    included_parts.c.sub_part,
    func.sum(included_parts.c.quantity).label("total_quantity"),
).group_by(included_parts.c.sub_part)

result = conn.execute(statement).fetchall()
```

### Example 3: Upsert using UPDATE and INSERT with CTEs

```python
from datetime import date
from sqlalchemy import (
    MetaData,
    Table,
    Column,
    Integer,
    Date,
    select,
    literal,
    and_,
    exists,
)

metadata = MetaData()

visitors = Table(
    "visitors",
    metadata,
    Column("product_id", Integer, primary_key=True),
    Column("date", Date, primary_key=True),
    Column("count", Integer),
)

# add 5 visitors for the product_id == 1
product_id = 1
day = date.today()
count = 5

update_cte = (
    visitors.update()
    .where(
        and_(visitors.c.product_id == product_id, visitors.c.date == day)
    )
    .values(count=visitors.c.count + count)
    .returning(literal(1))
    .cte("update_cte")
)

upsert = visitors.insert().from_select(
    [visitors.c.product_id, visitors.c.date, visitors.c.count],
    select(literal(product_id), literal(day), literal(count)).where(
        ~exists(update_cte.select())
    ),
)

connection.execute(upsert)
```

### Example 4: Nesting CTE (SQLAlchemy 1.4.24 and above)

```python
value_a = select(literal("root").label("n")).cte("value_a")

# A nested CTE with the same name as the root one
value_a_nested = select(literal("nesting").label("n")).cte(
    "value_a", nesting=True
)

# Nesting CTEs takes ascendency locally
# over the CTEs at a higher level
value_b = select(value_a_nested.c.n).cte("value_b")

value_ab = select(value_a.c.n.label("a"), value_b.c.n.label("b"))
```

The above query will render the second CTE nested inside the first, shown with inline parameters below as:

```sql
WITH
    value_a AS
        (SELECT 'root' AS n),
    value_b AS
        (WITH value_a AS
            (SELECT 'nesting' AS n)
        SELECT value_a.n AS n FROM value_a)
SELECT value_a.n AS a, value_b.n AS b
FROM value_a, value_b
```

```

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

### Organizing code with multiple Connection.begin() blocks

Source: https://docs.sqlalchemy.org/en/20/core/connections.html

Illustrates using Connection.begin() to organize code into separate context-managed transaction blocks within a single connection checkout.

```python
with engine.connect() as conn:
    with conn.begin():
        conn.execute(...)
        conn.execute(...)

    with conn.begin():
        conn.execute(...)
        conn.execute(...)
```

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

### FunctionFilter.over() longhand example

Source: https://docs.sqlalchemy.org/en/20/core/sqlelement.html

Shows the equivalent longhand expression using over() and funcfilter() to produce the same OVER clause as the shorthand example.

```python
from sqlalchemy import over, funcfilter

over(funcfilter(func.rank(), MyClass.y > 5), order_by="x")
```

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

### Index Creation with Column and Index Construct

Source: https://docs.sqlalchemy.org/en/20/core/constraints.html

Demonstrates creating indexes using the inline index keyword on Column and the Index construct for named and multi-column indexes.

```APIDOC
## Index Creation with Column and Index Construct

### Description
This example shows how to create indexes on a Table using the inline `index` keyword on Column and the `Index` construct for named and multi-column indexes.

### Usage
```python
metadata_obj = MetaData()
mytable = Table(
    "mytable",
    metadata_obj,
    # an indexed column, with index "ix_mytable_col1"
    Column("col1", Integer, index=True),
    # a uniquely indexed column with index "ix_mytable_col2"
    Column("col2", Integer, index=True, unique=True),
    Column("col3", Integer),
    Column("col4", Integer),
    Column("col5", Integer),
    Column("col6", Integer),
)

# place an index on col3, col4
Index("idx_col34", mytable.c.col3, mytable.c.col4)

# place a unique index on col5, col6
Index("myindex", mytable.c.col5, mytable.c.col6, unique=True)

mytable.create(engine)
```

### Generated DDL
```sql
CREATE TABLE mytable (
    col1 INTEGER,
    col2 INTEGER,
    col3 INTEGER,
    col4 INTEGER,
    col5 INTEGER,
    col6 INTEGER
)
CREATE INDEX ix_mytable_col1 ON mytable (col1)
CREATE UNIQUE INDEX ix_mytable_col2 ON mytable (col2)
CREATE UNIQUE INDEX myindex ON mytable (col5, col6)
CREATE INDEX idx_col34 ON mytable (col3, col4)
```
```

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

### TypeDecorator.__init__()

Source: https://docs.sqlalchemy.org/en/20/core/custom_types.html

Construct a TypeDecorator. This method initializes a new TypeDecorator instance.

```APIDOC
## TypeDecorator.__init__()

### Description
Construct a TypeDecorator. This method initializes a new TypeDecorator instance.

### Method
__init__()

### Parameters
- **&#42;args** - Positional arguments for the underlying type.
- **&#42;kwargs** - Keyword arguments for the underlying type.
```

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

### Using Connection.begin() with context manager

Source: https://docs.sqlalchemy.org/en/20/core/connections.html

Shows how to use Connection.begin() to explicitly begin a transaction within a connection context. The returned RootTransaction object works as a context manager and completes on commit or rollback.

```python
with engine.connect() as conn:
    with conn.begin() as trans:
        conn.execute(table.insert(), {"username": "sandy"})
```

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

### SQL generated by distinct() example

Source: https://docs.sqlalchemy.org/en/20/core/sqlelement.html

Shows the SQL statement produced by the distinct() example, with DISTINCT applied to the column expression.

```sql
SELECT user.id, count(DISTINCT user.name) FROM user
```

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

### Context-Sensitive Default Function Example

Source: https://docs.sqlalchemy.org/en/20/core/defaults.html

Example of a default function that uses the context to access other column values and compute a default.

```APIDOC
## Context-Sensitive Default Function

### Description
Define a Python function that accepts a `context` argument to compute a default value based on other values in the statement.

### Usage
Use the function as the `default` or `onupdate` parameter of a `Column`.

### Function Signature
- **mydefault(context)** - A Python function that takes a single `context` argument and returns a value.

### Context Methods
- **context.get_current_parameters()** - Returns a dictionary of column-key to values for the current INSERT or UPDATE statement.

### Example
```python
def mydefault(context):
    return context.get_current_parameters()["counter"] + 12

t = Table(
    "mytable",
    metadata_obj,
    Column("counter", Integer),
    Column("counter_plus_twelve", Integer, default=mydefault, onupdate=mydefault),
)
```

### Behavior
- The function is called for each set of parameters in executemany style.
- For multi-valued INSERT constructs, the function is called once per VALUES clause.
- The `get_current_parameters()` method returns the full set of values for the statement, or the subset for the individual VALUES clause in multi-valued INSERTs.
```

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

### URL.update_query_string example

Source: https://docs.sqlalchemy.org/en/20/core/engines.html

Example from the SQLAlchemy Engine Creation API documentation showing how to use URL.update_query_string to update the query parameters of a URL using a query string. The example creates a URL from a string, updates it with a query string, and prints the resulting URL string.

```python
>>> from sqlalchemy.engine import make_url
>>> url = make_url("postgresql+psycopg2://user:pass@host/dbname")
>>> url = url.update_query_string(
...     "alt_host=host1&alt_host=host2&ssl_cipher=%2Fpath%2Fto%2Fcrt"
... )
>>> str(url)
'postgresql+psycopg2://user:pass@host/dbname?alt_host=host1&alt_host=host2&ssl_cipher=%2Fpath%2Fto%2Fcrt'
```

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

### MetaData.__init__()

Source: https://docs.sqlalchemy.org/en/20/core/metadata.html

Constructor for MetaData.

```APIDOC
## MetaData.__init__()

### Description
Initializes a MetaData object.

### Method
__init__

### Parameters
- **schema** (string) - Optional - The schema name.
- **quote_schema** (bool) - Optional - Whether to quote the schema.
- **naming_convention** (dict) - Optional - Naming convention mapping.
- **info** (dict) - Optional - Additional info.

### Returns
None
```

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

### after_parent_attach event listener example

Source: https://docs.sqlalchemy.org/en/20/core/events.html

Example of listening for the 'after_parent_attach' event on a schema object. The listener receives target and parent arguments.

```python
from sqlalchemy import event


@event.listens_for(SomeSchemaClassOrObject, 'after_parent_attach')
def receive_after_parent_attach(target, parent):
    "listen for the 'after_parent_attach' event"

    # ... (event handling logic) ...
```

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

### Associating listeners with Engine and Pool targets

Source: https://docs.sqlalchemy.org/en/20/core/event.html

Shows how to use listen() to associate a 'connect' event with all Pool instances, all Engine instances, a specific pool, and a specific engine's pool. The example defines a connection function and creates a QueuePool and Engine to demonstrate the different target types.

```python
from sqlalchemy.event import listen
from sqlalchemy.pool import Pool, QueuePool
from sqlalchemy import create_engine
from sqlalchemy.engine import Engine
import psycopg2


def connect():
    return psycopg2.connect(user="ed", host="127.0.0.1", dbname="test")


my_pool = QueuePool(connect)
my_engine = create_engine("postgresql+psycopg2://ed@localhost/test")

# associate listener with all instances of Pool
listen(Pool, "connect", my_on_connect)

# associate listener with all instances of Pool
# via the Engine class
listen(Engine, "connect", my_on_connect)

# associate listener with my_pool
listen(my_pool, "connect", my_on_connect)

# associate listener with my_engine.pool
listen(my_engine, "connect", my_on_connect)
```

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

### columns attribute example

Source: https://docs.sqlalchemy.org/en/20/core/functions.html

Demonstrates the legacy use of FunctionElement.columns to place a function in the FROM clause. The example selects columns x and y from func.myfunction().

```python
>>> from sqlalchemy import column, select, func
>>> stmt = select(column("x"), column("y")).select_from(func.myfunction())
>>> print(stmt)
SELECT x, y FROM myfunction()
```

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

### CreateEnginePlugin.__init__()

Source: https://docs.sqlalchemy.org/en/20/core/connections.html

Constructs a new CreateEnginePlugin. This is typically called by the Engine creation process.

```APIDOC
## CreateEnginePlugin.__init__()

### Description
Constructs a new CreateEnginePlugin. This is typically called by the Engine creation process.

### Method
Constructor

### Signature
`CreateEnginePlugin.__init__(url, kwargs)`

### Parameters
- **url** (URL) - Required - The URL of the database.
- **kwargs** (dict) - Required - Keyword arguments passed to the plugin.
```

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

### Listen for 'prepare_twophase' event

Source: https://docs.sqlalchemy.org/en/20/core/events.html

Example argument form for the 'prepare_twophase' event. The listener receives the Connection object and the two-phase XID identifier.

```python
from sqlalchemy import event


@event.listens_for(SomeEngine, 'prepare_twophase')
def receive_prepare_twophase(conn, xid):
    "listen for the 'prepare_twophase' event"

    # ... (event handling logic) ...
```

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

### Begin Once Transaction with Connection.begin()

Source: https://docs.sqlalchemy.org/en/20/core/connections.html

Use this pattern to explicitly start a transaction and let the context manager handle commit/rollback. The transaction commits when the block exits normally, and rolls back if an exception is raised.

```python
with engine.connect() as connection:
    with connection.begin():
        connection.execute(some_table.insert(), {"x": 7, "y": "this is some data"})
        connection.execute(
            some_other_table.insert(), {"q": 8, "p": "this is some more data"}
        )

    # transaction is committed
```
Tokens:
Raw