SQLAlchemy

https://docs.sqlalchemy.org/en/20/
Admin
SQLAlchemy is a comprehensive database toolkit for Python, providing both a full-featured Object...
Tokens:3,820,752
Snippets:29,001
Trust Score:9.7
Update:2 days ago
Show doc for...
Tokens:
Raw
### Setup imports and Base class for relationship examples

Source: https://docs.sqlalchemy.org/en/20/orm/basic_relationships.html

Common setup for all examples: imports and a DeclarativeBase subclass named Base.

```python
from __future__ import annotations
from typing import List

from sqlalchemy import ForeignKey
from sqlalchemy import Integer
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import relationship


class Base(DeclarativeBase):
    pass
```

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

### Define ORM Models and Setup Session for DML Examples

Source: https://docs.sqlalchemy.org/en/20/orm/queryguide/_dml_setup.html

Defines the Base, User, Address, LogRecord, and Employee/Manager/Engineer models, then creates an in-memory SQLite engine, creates tables, and opens a session with a transaction. Use this as the prerequisite setup for the ORM DML examples.

```python
>>> from typing import List
>>> from typing import Optional
>>> import datetime
>>>
>>> from sqlalchemy import Column
>>> from sqlalchemy import create_engine
>>> from sqlalchemy import ForeignKey
>>> from sqlalchemy import Table
>>> from sqlalchemy.orm import DeclarativeBase
>>> from sqlalchemy.orm import Mapped
>>> from sqlalchemy.orm import mapped_column
>>> from sqlalchemy.orm import relationship
>>> from sqlalchemy.orm import Session
>>>
>>>
>>> class Base(DeclarativeBase):
...     pass
>>> class User(Base):
...     __tablename__ = "user_account"
...     id: Mapped[int] = mapped_column(primary_key=True)
...     name: Mapped[str] = mapped_column(unique=True)
...     fullname: Mapped[Optional[str]]
...     species: Mapped[Optional[str]]
...     addresses: Mapped[List["Address"]] = relationship(back_populates="user")
...
...     def __repr__(self) -> str:
...         return f"User(name={self.name!r}, fullname={self.fullname!r})"
>>> class Address(Base):
...     __tablename__ = "address"
...     id: Mapped[int] = mapped_column(primary_key=True)
...     user_id: Mapped[int] = mapped_column(ForeignKey("user_account.id"))
...     email_address: Mapped[str]
...     user: Mapped[User] = relationship(back_populates="addresses")
...
...     def __repr__(self) -> str:
...         return f"Address(email_address={self.email_address!r})"
>>> class LogRecord(Base):
...     __tablename__ = "log_record"
...     id: Mapped[int] = mapped_column(primary_key=True)
...     message: Mapped[str]
...     code: Mapped[str]
...     timestamp: Mapped[datetime.datetime]
...
...     def __repr__(self):
...         return f"LogRecord({self.message!r}, {self.code!r}, {self.timestamp!r})"

>>> class Employee(Base):
...     __tablename__ = "employee"
...     id: Mapped[int] = mapped_column(primary_key=True)
...     name: Mapped[str]
...     type: Mapped[str]
...
...     def __repr__(self):
...         return f"{self.__class__.__name__}({self.name!r})"
...
...     __mapper_args__ = {
...         "polymorphic_identity": "employee",
...         "polymorphic_on": "type",
...     }
>>> class Manager(Employee):
...     __tablename__ = "manager"
...     id: Mapped[int] = mapped_column(ForeignKey("employee.id"), primary_key=True)
...     manager_name: Mapped[str]
...
...     def __repr__(self):
...         return f"{self.__class__.__name__}({self.name!r}, manager_name={self.manager_name!r})"
...
...     __mapper_args__ = {
...         "polymorphic_identity": "manager",
...     }
>>> class Engineer(Employee):
...     __tablename__ = "engineer"
...     id: Mapped[int] = mapped_column(ForeignKey("employee.id"), primary_key=True)
...     engineer_info: Mapped[str]
...
...     def __repr__(self):
...         return f"{self.__class__.__name__}({self.name!r}, engineer_info={self.engineer_info!r})"
...
...     __mapper_args__ = {
...         "polymorphic_identity": "engineer",
...     }

>>> engine = create_engine("sqlite+pysqlite:///:memory:", echo=True)
>>> Base.metadata.create_all(engine)
BEGIN ...
>>> conn = engine.connect()
>>> session = Session(conn)
>>> conn.begin()
BEGIN ...
```

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

### Sharding with asyncio example

Source: https://docs.sqlalchemy.org/en/20/_modules/examples/sharding/asyncio.html

Full source code for examples.sharding.asyncio, showing async sharding setup with ShardedSession, async_sessionmaker, and shard configuration.

```python
"""Illustrates sharding API used with asyncio.

For the sync version of this example, see separate_databases.py.

Most of the code here is copied from separate_databases.py and works
in exactly the same way.   The main change is how the
``async_sessionmaker`` is configured, and as is specific to this example
the routine that generates new primary keys.

"""

from __future__ import annotations

import asyncio
import datetime

from sqlalchemy import Column
from sqlalchemy import ForeignKey
from sqlalchemy import inspect
from sqlalchemy import Integer
from sqlalchemy import select
from sqlalchemy import Table
from sqlalchemy.ext.asyncio import async_sessionmaker
from sqlalchemy.ext.asyncio import create_async_engine
from sqlalchemy.ext.horizontal_shard import set_shard_id
from sqlalchemy.ext.horizontal_shard import ShardedSession
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import immediateload
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column
from sqlalchemy.orm import relationship
from sqlalchemy.sql import operators
from sqlalchemy.sql import visitors

echo = True
db1 = create_async_engine("sqlite+aiosqlite://", echo=echo)
db2 = create_async_engine("sqlite+aiosqlite://", echo=echo)
db3 = create_async_engine("sqlite+aiosqlite://", echo=echo)
db4 = create_async_engine("sqlite+aiosqlite://", echo=echo)


# for asyncio, the ShardedSession class is passed
# via sync_session_class.  The shards themselves are used within
# implicit-awaited internals, so we use the sync_engine Engine objects
# in the shards dictionary.
Session = async_sessionmaker(
    sync_session_class=ShardedSession,
    expire_on_commit=False,
    shards={
        "north_america": db1.sync_engine,
        "asia": db2.sync_engine,
        "europe": db3.sync_engine,
        "south_america": db4.sync_engine,
    },
)


# mappings and tables
class Base(DeclarativeBase):
    pass


# we need a way to create identifiers which are unique across all databases.
# one easy way would be to just use a composite primary key, where  one value
# is the shard id.  but here, we'll show something more "generic", an id
# generation function.  we'll use a simplistic "id table" stored in database
# #1.  Any other method will do just as well; UUID, hilo, application-specific,
# etc.

ids = Table("ids", Base.metadata, Column("nextid", Integer, nullable=False))


def id_generator(ctx):
    # id_generator is run within a "synchronous" context, where
    # we use an implicit-await API that will convert back to explicit await
    # calls when it reaches the driver.
    with db1.sync_engine.begin() as conn:
        nextid = conn.scalar(ids.select().with_for_update())
        conn.execute(ids.update().values({ids.c.nextid: ids.c.nextid + 1}))
    return nextid


# table setup.  we'll store a lead table of continents/cities, and a secondary
# table storing locations. a particular row will be placed in the database
# whose shard id corresponds to the 'continent'.  in this setup, secondary rows
# in 'weather_reports' will be placed in the same DB as that of the parent, but
# this can be changed if you're willing to write more complex sharding
# functions.


class WeatherLocation(Base):
    __tablename__ = "weather_locations"

    id: Mapped[int] = mapped_column(primary_key=True, default=id_generator)
    continent: Mapped[str]
    city: Mapped[str]

    reports: Mapped[list[Report]] = relationship(back_populates="location")

    def __init__(self, continent: str, city: str):
        self.continent = continent
        self.city = city


class Report(Base):
    __tablename__ = "weather_reports"

    id: Mapped[int] = mapped_column(primary_key=True)
    location_id: Mapped[int] = mapped_column(
        ForeignKey("weather_locations.id")
    )
    temperature: Mapped[float]
    report_time: Mapped[datetime.datetime] = mapped_column(
        default=datetime.datetime.now
    )

    location: Mapped[WeatherLocation] = relationship(back_populates="reports")

    def __init__(self, temperature: float):
        self.temperature = temperature


# step 5. define sharding functions.

# we'll use a straight mapping of a particular set of "country"
# attributes to shard id.
shard_lookup = {
    "North America": "north_america",
    "Asia": "asia",
    "Europe": "europe",
    "South America": "south_america",
}


def shard_chooser(mapper, instance, clause=None):
    """shard chooser.

```

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

### Install sample data with install() in examples.dogpile_caching.fixture_data

Source: https://docs.sqlalchemy.org/en/20/_modules/examples/dogpile_caching/fixture_data.html

Creates all tables via Base.metadata.create_all and populates cities, countries, postal codes, and 50 Person records with random postal codes. Commits the session and calls Session.remove() to start the demo fresh.

```python
"""Installs some sample data.   Here we have a handful of postal codes for
a few US/Canadian cities.   Then, 100 Person records are installed, each
with a randomly selected postal code.

"""

import random

from .environment import Base
from .environment import Session
from .model import Address
from .model import City
from .model import Country
from .model import Person
from .model import PostalCode


def install():
    Base.metadata.create_all(Session().bind)

    data = [
        ("Chicago", "United States", ("60601", "60602", "60603", "60604")),
        ("Montreal", "Canada", ("H2S 3K9", "H2B 1V4", "H7G 2T8")),
        ("Edmonton", "Canada", ("T5J 1R9", "T5J 1Z4", "T5H 1P6")),
        (
            "New York",
            "United States",
            ("10001", "10002", "10003", "10004", "10005", "10006"),
        ),
        (
            "San Francisco",
            "United States",
            ("94102", "94103", "94104", "94105", "94107", "94108"),
        ),
    ]

    countries = {}
    all_post_codes = []
    for city, country, postcodes in data:
        try:
            country = countries[country]
        except KeyError:
            countries[country] = country = Country(country)

        city = City(city, country)
        pc = [PostalCode(code, city) for code in postcodes]
        Session.add_all(pc)
        all_post_codes.extend(pc)

    for i in range(1, 51):
        person = Person(
            "person %.2d" % i,
            Address(
                street="street %.2d" % i,
                postal_code=all_post_codes[
                    random.randint(0, len(all_post_codes) - 1)
                ],
            ),
        )
        Session.add(person)

    Session.commit()

    # start the demo fresh
    Session.remove()

```

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

### ORM Query Guide SELECT Setup: Models and Fixture Data

Source: https://docs.sqlalchemy.org/en/20/orm/queryguide/_plain_setup.html

Defines the User, Address, Order, and Item mapped classes with relationships, creates an in-memory SQLite database, and inserts sample data. Use this as the foundation for the SELECT examples in the ORM Querying Guide.

```python
>>> from typing import List
>>> from typing import Optional
>>>
>>> from sqlalchemy import Column
>>> from sqlalchemy import create_engine
>>> from sqlalchemy import ForeignKey
>>> from sqlalchemy import Table
>>> from sqlalchemy.orm import DeclarativeBase
>>> from sqlalchemy.orm import Mapped
>>> from sqlalchemy.orm import mapped_column
>>> from sqlalchemy.orm import relationship
>>> from sqlalchemy.orm import Session
>>>
>>>
>>> class Base(DeclarativeBase):
...     pass
>>> class User(Base):
...     __tablename__ = "user_account"
...     id: Mapped[int] = mapped_column(primary_key=True)
...     name: Mapped[str]
...     fullname: Mapped[Optional[str]]
...     addresses: Mapped[List["Address"]] = relationship(back_populates="user")
...     orders: Mapped[List["Order"]] = relationship()
...
...     def __repr__(self) -> str:
...         return f"User(id={self.id!r}, name={self.name!r}, fullname={self.fullname!r})"
>>> class Address(Base):
...     __tablename__ = "address"
...     id: Mapped[int] = mapped_column(primary_key=True)
...     user_id: Mapped[int] = mapped_column(ForeignKey("user_account.id"))
...     email_address: Mapped[str]
...     user: Mapped[User] = relationship(back_populates="addresses")
...
...     def __repr__(self) -> str:
...         return f"Address(id={self.id!r}, email_address={self.email_address!r})"
>>> order_items_table = Table(
...     "order_items",
...     Base.metadata,
...     Column("order_id", ForeignKey("user_order.id"), primary_key=True),
...     Column("item_id", ForeignKey("item.id"), primary_key=True),
... )
>>>
>>> class Order(Base):
...     __tablename__ = "user_order"
...     id: Mapped[int] = mapped_column(primary_key=True)
...     user_id: Mapped[int] = mapped_column(ForeignKey("user_account.id"))
...     items: Mapped[List["Item"]] = relationship(secondary=order_items_table)
>>> class Item(Base):
...     __tablename__ = "item"
...     id: Mapped[int] = mapped_column(primary_key=True)
...     name: Mapped[str]
...     description: Mapped[str]
>>> engine = create_engine("sqlite+pysqlite:///:memory:", echo=True)
>>> Base.metadata.create_all(engine)
BEGIN ...
>>> conn = engine.connect()
>>> session = Session(conn)
>>> session.add_all(
...     [
...         User(
...             name="spongebob",
...             fullname="Spongebob Squarepants",
...             addresses=[Address(email_address="[email protected]")],
...         ),
...         User(
...             name="sandy",
...             fullname="Sandy Cheeks",
...             addresses=[
...                 Address(email_address="[email protected]"),
...                 Address(email_address="[email protected]"),
...             ],
...         ),
...         User(
...             name="patrick",
...             fullname="Patrick Star",
...             addresses=[Address(email_address="[email protected]")],
...         ),
...         User(
...             name="squidward",
...             fullname="Squidward Tentacles",
...             addresses=[Address(email_address="[email protected]")],
...         ),
...         User(name="ehkrabs", fullname="Eugene H. Krabs"),
...     ]
... )
>>> session.commit()
BEGIN ... COMMIT
>>> conn.begin()
BEGIN ...
```

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

### Create a Session instance

Source: https://docs.sqlalchemy.org/en/20/orm/session_transaction.html

Initializes a Session bound to an engine. This is the starting point for the transaction examples.

```python
from sqlalchemy.orm import Session

session = Session(engine)
```

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

### 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

```

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

### setup_database creates engine and bulk-inserts Customer rows

Source: https://docs.sqlalchemy.org/en/20/_modules/examples/performance/large_resultsets.html

Decorated with @Profiler.setup_once, this drops and recreates tables, then inserts rows in chunks of 10000 using Core insert with a params list. It commits after all chunks are inserted.

```python
@Profiler.setup_once
def setup_database(dburl, echo, num):
    global engine
    engine = create_engine(dburl, echo=echo)
    Base.metadata.drop_all(engine)
    Base.metadata.create_all(engine)

    s = Session(engine)
    for chunk in range(0, num, 10000):
        s.execute(
            Customer.__table__.insert(),
            params=[
                {
                    "name": "customer name %d" % i,
                    "description": "customer description %d" % i,
                }
                for i in range(chunk, chunk + 10000)
            ],
        )
    s.commit()
```

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

### Define joined inheritance mappings and insert fixture data

Source: https://docs.sqlalchemy.org/en/20/orm/queryguide/_inheritance_setup.html

Defines the Company, Employee, Manager, Engineer, and Paperwork classes using joined table inheritance, creates an in-memory SQLite database, and inserts sample data. Use this setup to follow the inheritance query examples in the ORM Querying Guide.

```python
>>> from typing import List
>>> from sqlalchemy import create_engine
>>> from sqlalchemy import ForeignKey
>>> from sqlalchemy.orm import DeclarativeBase
>>> from sqlalchemy.orm import Mapped
>>> from sqlalchemy.orm import mapped_column
>>> from sqlalchemy.orm import relationship
>>> from sqlalchemy.orm import Session
>>>
>>>
>>> class Base(DeclarativeBase):
...     pass
>>> class Company(Base):
...     __tablename__ = "company"
...     id: Mapped[int] = mapped_column(primary_key=True)
...     name: Mapped[str]
...     employees: Mapped[List["Employee"]] = relationship(back_populates="company")
>>>
>>> class Employee(Base):
...     __tablename__ = "employee"
...     id: Mapped[int] = mapped_column(primary_key=True)
...     name: Mapped[str]
...     type: Mapped[str]
...     company_id: Mapped[int] = mapped_column(ForeignKey("company.id"))
...     company: Mapped[Company] = relationship(back_populates="employees")
...
...     def __repr__(self):
...         return f"{self.__class__.__name__}({self.name!r})"
...
...     __mapper_args__ = {
...         "polymorphic_identity": "employee",
...         "polymorphic_on": "type",
...     }
>>>
>>> class Manager(Employee):
...     __tablename__ = "manager"
...     id: Mapped[int] = mapped_column(ForeignKey("employee.id"), primary_key=True)
...     manager_name: Mapped[str]
...     paperwork: Mapped[List["Paperwork"]] = relationship()
...     __mapper_args__ = {
...         "polymorphic_identity": "manager",
...     }
>>> class Paperwork(Base):
...     __tablename__ = "paperwork"
...     id: Mapped[int] = mapped_column(primary_key=True)
...     manager_id: Mapped[int] = mapped_column(ForeignKey("manager.id"))
...     document_name: Mapped[str]
...
...     def __repr__(self):
...         return f"Paperwork({self.document_name!r})"
>>>
>>> class Engineer(Employee):
...     __tablename__ = "engineer"
...     id: Mapped[int] = mapped_column(ForeignKey("employee.id"), primary_key=True)
...     engineer_info: Mapped[str]
...     __mapper_args__ = {
...         "polymorphic_identity": "engineer",
...     }
>>>
>>> engine = create_engine("sqlite://", echo=True)
>>>
>>> Base.metadata.create_all(engine)
BEGIN ...

>>> conn = engine.connect()
>>> from sqlalchemy.orm import Session
>>> session = Session(conn)
>>> session.add(
...     Company(
...         name="Krusty Krab",
...         employees=[
...             Manager(
...                 name="Mr. Krabs",
...                 manager_name="Eugene H. Krabs",
...                 paperwork=[
...                     Paperwork(document_name="Secret Recipes"),
...                     Paperwork(document_name="Krabby Patty Orders"),
...                 ],
...             ),
...             Engineer(name="SpongeBob", engineer_info="Krabby Patty Master"),
...             Engineer(
...                 name="Squidward",
...                 engineer_info="Senior Customer Engagement Engineer",
...             ),
...         ],
...     )
... )
>>> session.commit()
BEGIN ...
```

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

### Run performance suite help

Source: https://docs.sqlalchemy.org/en/20/orm/examples.html

Displays the command-line help for the examples.performance package, showing available suites and options.

```bash
$ python -m examples.performance --help
usage: python -m examples.performance [-h] [--test TEST] [--dburl DBURL]
                                      [--num NUM] [--profile] [--dump]
                                      [--echo]

                                      {bulk_inserts,large_resultsets,single_inserts}

positional arguments:
  {bulk_inserts,large_resultsets,single_inserts}
                        suite to run

optional arguments:
  -h, --help            show this help message and exit
  --test TEST           run specific test name
  --dburl DBURL         database URL, default sqlite:///profile.db
  --num NUM             Number of iterations/items/etc for tests;
                        default is module-specific
  --profile             run profiling and dump call counts
  --dump                dump full call profile (implies --profile)
  --echo                Echo SQL output
```

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

### 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"))
```

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

### Async main function to save and load objects

Source: https://docs.sqlalchemy.org/en/20/_modules/examples/sharding/asyncio.html

Calls setup, then creates WeatherLocation and Report objects and appends reports to demonstrate sharding behavior.

```python
async def main():
    await setup()

    # save and load objects!

    tokyo = WeatherLocation("Asia", "Tokyo")
    newyork = WeatherLocation("North America", "New York")
    toronto = WeatherLocation("North America", "Toronto")
    london = WeatherLocation("Europe", "London")
    dublin = WeatherLocation("Europe", "Dublin")
    brasilia = WeatherLocation("South America", "Brasila")
    quito = WeatherLocation("South America", "Quito")

    tokyo.reports.append(Report(80.0))
    newyork.reports.append(Report(75))
    quito.reports.append(Report(85))
```

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

### Space Invaders Example Source Code

Source: https://docs.sqlalchemy.org/en/20/_modules/examples/space_invaders/space_invaders.html

Complete source code for the Space Invaders example, including imports, logging setup, constants, and ORM model definitions for Glyph and GlyphCoordinate.

```python
import curses
import logging
import random
import re
import textwrap
import time

from sqlalchemy import Column
from sqlalchemy import create_engine
from sqlalchemy import ForeignKey
from sqlalchemy import func
from sqlalchemy import Integer
from sqlalchemy import String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.ext.hybrid import hybrid_method
from sqlalchemy.ext.hybrid import hybrid_property
from sqlalchemy.orm import joinedload
from sqlalchemy.orm import relationship
from sqlalchemy.orm import Session

logging.basicConfig(
    filename="space_invaders.log",
    format="%(asctime)s,%(msecs)03d %(levelname)-5.5s %(message)s",
)
logging.getLogger("sqlalchemy.engine").setLevel(logging.INFO)

Base = declarative_base()

WINDOW_LEFT = 10
WINDOW_TOP = 2
WINDOW_WIDTH = 70
WINDOW_HEIGHT = 34
VERT_PADDING = 2
HORIZ_PADDING = 5
ENEMY_VERT_SPACING = 4
MAX_X = WINDOW_WIDTH - HORIZ_PADDING
MAX_Y = WINDOW_HEIGHT - VERT_PADDING
LEFT_KEY = ord("j")
RIGHT_KEY = ord("l")
FIRE_KEY = ord(" ")
PAUSE_KEY = ord("p")

COLOR_MAP = {
    "K": curses.COLOR_BLACK,
    "B": curses.COLOR_BLUE,
    "C": curses.COLOR_CYAN,
    "G": curses.COLOR_GREEN,
    "M": curses.COLOR_MAGENTA,
    "R": curses.COLOR_RED,
    "W": curses.COLOR_WHITE,
    "Y": curses.COLOR_YELLOW,
}


class Glyph(Base):
    """Describe a "glyph", a graphical element
    to be painted on the screen.

    """

    __tablename__ = "glyph"
    id = Column(Integer, primary_key=True)
    name = Column(String)
    type = Column(String)
    width = Column(Integer)
    height = Column(Integer)
    data = Column(String)
    alt_data = Column(String)
    __mapper_args__ = {"polymorphic_on": type}

    def __init__(self, name, img, alt=None):
        self.name = name
        self.data, self.width, self.height = self._encode_glyph(img)
        if alt is not None:
            self.alt_data, alt_w, alt_h = self._encode_glyph(alt)

    def _encode_glyph(self, img):
        """Receive a textual description of the glyph and
        encode into a format understood by
        GlyphCoordinate.render().

        """
        img = re.sub(r"^\n", "", textwrap.dedent(img))
        color = "W"
        lines = [line.rstrip() for line in img.split("\n")]
        data = []
        for line in lines:
            render_line = []
            line = list(line)
            while line:
                char = line.pop(0)
                if char == "#":
                    color = line.pop(0)
                    continue
                render_line.append((color, char))
            data.append(render_line)
        width = max([len(rl) for rl in data])
        data = "".join(
            "".join("%s%s" % (color, char) for color, char in render_line)
            + ("W " * (width - len(render_line)))
            for render_line in data
        )
        return data, width, len(lines)

    def glyph_for_state(self, coord, state):
        """Return the appropriate data representation
        for this Glyph, based on the current coordinates
        and state.

        Subclasses may override this to provide animations.

        """
        return self.data


class GlyphCoordinate(Base):
    """Describe a glyph rendered at a certain x, y coordinate.

    The GlyphCoordinate may also include optional values
    such as the tick at time of render, a label, and a
    score value.

    """

    __tablename__ = "glyph_coordinate"
    id = Column(Integer, primary_key=True)
    glyph_id = Column(Integer, ForeignKey("glyph.id"))
    x = Column(Integer)
    y = Column(Integer)
    tick = Column(Integer)
    label = Column(String)
    score = Column(Integer)
    glyph = relationship(Glyph, innerjoin=True)

    def __init__(
        self, session, glyph_name, x, y, tick=None, label=None, score=None
    ):
        self.glyph = session.query(Glyph).filter_by(name=glyph_name).one()
        self.x = x
        self.y = y
        self.tick = tick
        self.label = label
        self.score = score
        session.add(self)

    def render(self, window, state):
        """Render the Glyph at this position."""

```

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

### Async Core Example with create_async_engine and run_sync

Source: https://docs.sqlalchemy.org/en/20/orm/extensions/asyncio.html

Complete example of using AsyncEngine with SQLite. Shows creating tables via run_sync, inserting rows, selecting data, and disposing the engine. The output includes SQL logging.

```python
>>> import asyncio

>>> from sqlalchemy import Column
>>> from sqlalchemy import MetaData
>>> from sqlalchemy import select
>>> from sqlalchemy import String
>>> from sqlalchemy import Table
>>> from sqlalchemy.ext.asyncio import create_async_engine

>>> meta = MetaData()
>>> t1 = Table("t1", meta, Column("name", String(50), primary_key=True))


>>> async def async_main() -> None:
...     engine = create_async_engine("sqlite+aiosqlite://", echo=True)
...
...     async with engine.begin() as conn:
...         await conn.run_sync(meta.drop_all)
...         await conn.run_sync(meta.create_all)
...
...         await conn.execute(
...             t1.insert(), [{"name": "some name 1"}, {"name": "some name 2"}]
...         )
...
...     async with engine.connect() as conn:
...         # select a Result, which will be delivered with buffered
...         # results
...         result = await conn.execute(select(t1).where(t1.c.name == "some name 1"))
...
...         print(result.fetchall())
...
...     # for AsyncEngine created in function scope, close and
...     # clean-up pooled connections
...     await engine.dispose()


>>> asyncio.run(async_main())
BEGIN (implicit)
...
CREATE TABLE t1 (
    name VARCHAR(50) NOT NULL,
    PRIMARY KEY (name)
)
...
INSERT INTO t1 (name) VALUES (?)
[...] [('some name 1',), ('some name 2',)]
COMMIT
BEGIN (implicit)
SELECT t1.name
FROM t1
WHERE t1.name = ?
[...] ('some name 1',)
[('some name 1',)]
ROLLBACK
```

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

### 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

```

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

### Run single_inserts performance benchmark with Profiler

Source: https://docs.sqlalchemy.org/en/20/_modules/examples/performance/single_inserts.html

Full example module for examples.performance.single_inserts. Profiler.init("single_inserts", num=10000) sets the iteration count; setup_database drops and recreates the customer table and swaps in a StaticPool for SQLite. Run via Profiler.main() under __main__.

```python
"""In this series of tests, we're looking at a method that inserts a row
within a distinct transaction, and afterwards returns to essentially a
"closed" state.   This would be analogous to an API call that starts up
a database connection, inserts the row, commits and closes.

"""

from sqlalchemy import bindparam
from sqlalchemy import Column
from sqlalchemy import create_engine
from sqlalchemy import Identity
from sqlalchemy import Integer
from sqlalchemy import pool
from sqlalchemy import String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import Session
from . import Profiler

Base = declarative_base()
engine = None


class Customer(Base):
    __tablename__ = "customer"
    id = Column(Integer, Identity(), primary_key=True)
    name = Column(String(255))
    description = Column(String(255))


Profiler.init("single_inserts", num=10000)


@Profiler.setup
def setup_database(dburl, echo, num):
    global engine
    engine = create_engine(dburl, echo=echo)
    if engine.dialect.name == "sqlite":
        engine.pool = pool.StaticPool(creator=engine.pool._creator)
    Base.metadata.drop_all(engine)
    Base.metadata.create_all(engine)


@Profiler.profile
def test_orm_commit(n):
    """Individual INSERT/COMMIT pairs via the ORM"""

    for i in range(n):
        session = Session(bind=engine)
        session.add(
            Customer(
                name="customer name %d" % i,
                description="customer description %d" % i,
            )
        )
        session.commit()


@Profiler.profile
def test_bulk_save(n):
    """Individual INSERT/COMMIT pairs using the "bulk" API"""

    for i in range(n):
        session = Session(bind=engine)
        session.bulk_save_objects(
            [
                Customer(
                    name="customer name %d" % i,
                    description="customer description %d" % i,
                )
            ]
        )
        session.commit()


@Profiler.profile
def test_bulk_insert_dictionaries(n):
    """Individual INSERT/COMMIT pairs using the "bulk" API with dictionaries"""

    for i in range(n):
        session = Session(bind=engine)
        session.bulk_insert_mappings(
            Customer,
            [
                dict(
                    name="customer name %d" % i,
                    description="customer description %d" % i,
                )
            ],
        )
        session.commit()


@Profiler.profile
def test_core(n):
    """Individual INSERT/COMMIT pairs using Core."""

    for i in range(n):
        with engine.begin() as conn:
            conn.execute(
                Customer.__table__.insert(),
                dict(
                    name="customer name %d" % i,
                    description="customer description %d" % i,
                ),
            )


@Profiler.profile
def test_core_query_caching(n):
    """Individual INSERT/COMMIT pairs using Core with query caching"""

    cache = {}
    ins = Customer.__table__.insert()
    for i in range(n):
        with engine.begin() as conn:
            conn.execution_options(compiled_cache=cache).execute(
                ins,
                dict(
                    name="customer name %d" % i,
                    description="customer description %d" % i,
                ),
            )


@Profiler.profile
def test_dbapi_raw_w_connect(n):
    """Individual INSERT/COMMIT pairs w/ DBAPI + connection each time"""

    _test_dbapi_raw(n, True)


@Profiler.profile
def test_dbapi_raw_w_pool(n):
    """Individual INSERT/COMMIT pairs w/ DBAPI + connection pool"""

    _test_dbapi_raw(n, False)


def _test_dbapi_raw(n, connect):
    compiled = (
        Customer.__table__.insert()
        .values(name=bindparam("name"), description=bindparam("description"))
        .compile(dialect=engine.dialect)
    )

    if compiled.positional:
        args = (
            ("customer name %d" % i, "customer description %d" % i)
            for i in range(n)
        )
    else:
        args = (
            dict(
                name="customer name %d" % i,
                description="customer description %d" % i,
            )
            for i in range(n)
        )
    sql = str(compiled)

    if connect:
        for arg in args:
            # there's no connection pool, so if these were distinct
            # calls, we'd be connecting each time
            conn = engine.pool._creator()
            cursor = conn.cursor()
            cursor.execute(sql, arg)
            cursor.lastrowid
            conn.commit()
            conn.close()
    else:
        for arg in args:
            conn = engine.raw_connection()
            cursor = conn.cursor()
            cursor.execute(sql, arg)
            cursor.lastrowid
            conn.commit()
            conn.close()


if __name__ == "__main__":
    Profiler.main()

```

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

### Listen for the 'before_flush' event

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

Example argument forms for the SessionEvents.before_flush event listener. Use this to execute logic before the flush process starts.

```python
from sqlalchemy import event


@event.listens_for(SomeSessionClassOrObject, 'before_flush')
def receive_before_flush(session, flush_context, instances):
    "listen for the 'before_flush' event"

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

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

### Async setup function to create tables and initialize id

Source: https://docs.sqlalchemy.org/en/20/_modules/examples/sharding/asyncio.html

Creates all tables in each shard database and inserts the initial id value into db1.

```python
async def setup():
    # create tables
    for db in (db1, db2, db3, db4):
        async with db.begin() as conn:
            await conn.run_sync(Base.metadata.create_all)

    # establish initial "id" in db1
    async with db1.begin() as conn:
        await conn.execute(ids.insert(), {"nextid": 1})
```

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

### 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"))
```

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

### Legacy Query usage examples

Source: https://docs.sqlalchemy.org/en/20/changelog/migration_20.html

Illustrates various legacy Query methods: one(), first(), get(), join(), options(), from_statement().

```python
session = Session(engine)

# becomes legacy use case
user = session.query(User).filter_by(name="some user").one()

# becomes legacy use case
user = session.query(User).filter_by(name="some user").first()

# becomes legacy use case
user = session.query(User).get(5)

# becomes legacy use case
for user in (
    session.query(User).join(User.addresses).filter(Address.email == "[email protected]")
):
    ...

# becomes legacy use case
users = session.query(User).options(joinedload(User.addresses)).order_by(User.id).all()

# becomes legacy use case
users = session.query(User).from_statement(text("select * from users")).all()

# etc
```

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

### Create tables across all shards

Source: https://docs.sqlalchemy.org/en/20/_modules/examples/sharding/separate_schema_translates.html

The setup function creates all tables defined in Base.metadata on each of the four databases (db1, db2, db3, db4). This must be called before using the session.

```python
def setup():
    # create tables
    for db in (db1, db2, db3, db4):
        Base.metadata.create_all(db)
```

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

### Polymorphic mapping with column_property

Source: https://docs.sqlalchemy.org/en/20/changelog/migration_10.html

Mapping example with polymorphic loading enabled. This setup is used to demonstrate order_by label issues with polymorphic loading.

```python
class A(Base):
    __tablename__ = "a"

    id = Column(Integer, primary_key=True)
    type = Column(String)

    __mapper_args__ = {"polymorphic_on": type, "with_polymorphic": "*"}
```

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

### Inspect an AliasedClass with inspect()

Source: https://docs.sqlalchemy.org/en/20/orm/queryguide/api.html

Use inspect() on an AliasedClass to get an AliasedInsp object. The example assumes MyClass is already mapped and aliased is imported.

```python
from sqlalchemy import inspect

my_alias = aliased(MyClass)
insp = inspect(my_alias)
```

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

### Install SQLAlchemy from source distribution using setup.py

Source: https://docs.sqlalchemy.org/en/20/intro.html

Installs SQLAlchemy from the source distribution using the setup.py script. This method is platform agnostic and will attempt to build Cython extensions if possible, falling back to pure Python otherwise.

```shell
python setup.py install
```

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

### Define column_property with correlated subquery

Source: https://docs.sqlalchemy.org/en/20/changelog/migration_10.html

Mapping example where A.b is a column_property using a correlated subquery. This setup is used to demonstrate the order_by issues with aliases.

```python
class A(Base):
    __tablename__ = "a"

    id = Column(Integer, primary_key=True)


class B(Base):
    __tablename__ = "b"

    id = Column(Integer, primary_key=True)
    a_id = Column(ForeignKey("a.id"))


A.b = column_property(select([func.max(B.id)]).where(B.a_id == A.id).correlate(A))
```

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

### 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

```

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

### 1.x style mapping using Column

Source: https://docs.sqlalchemy.org/en/20/changelog/whatsnew_20.html

Example of a 1.x style declarative mapping using Column and relationship. This serves as the starting point for replacing Column with mapped_column().

```python
from sqlalchemy import Column
from sqlalchemy.orm import relationship
from sqlalchemy.orm import DeclarativeBase


class Base(DeclarativeBase):
    pass


class User(Base):
    __tablename__ = "user_account"

    id = Column(Integer, primary_key=True)
    name = Column(String(30), nullable=False)
    fullname = Column(String)
    addresses = relationship("Address", back_populates="user")


class Address(Base):
    __tablename__ = "address"

    id = Column(Integer, primary_key=True)
    email_address = Column(String, nullable=False)
    user_id = Column(ForeignKey("user_account.id"), nullable=False)
    user = relationship("User", back_populates="addresses")
```

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

### Run performance examples package as a script via Profiler.main()

Source: https://docs.sqlalchemy.org/en/20/_modules/examples/performance/__main__.html

Place in examples/performance/__main__.py so the package can be executed as a script. Imports Profiler from the package and calls Profiler.main() only when run as __main__.

```python
"""Allows the examples/performance package to be run as a script."""

from . import Profiler

if __name__ == "__main__":
    Profiler.main()
```

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

### Example output of inline parameter rendering with qmark style

Source: https://docs.sqlalchemy.org/en/20/faq/sqlexpressions.html

Shows the SQL string produced by the qmark rendering example, with parameters substituted in positional order.

```sql
SELECT a.id, a.data
FROM a
WHERE a.data = UUID('1bd70375-db17-4d8c-94f1-fc2ef3aada26')
```

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

### Create a Foo instance with no state

Source: https://docs.sqlalchemy.org/en/20/changelog/migration_10.html

Creates an instance of Foo with no attribute values set. Used as the starting point for examples demonstrating attribute access behavior.

```python
>>> obj = Foo()
```

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

### 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")
```

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

### Scalar Results with Result.scalars()

Source: https://docs.sqlalchemy.org/en/20/changelog/whatsnew_20.html

Use Result.scalars() to get scalar values from a single-column statement. The example shows executing a select and retrieving all scalar results as a sequence.

```python
# (variable) data: Sequence[str]
data = connection.execute(select(str_col)).scalars().all()
```

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

### Rendered CREATE TABLE with partitioning

Source: https://docs.sqlalchemy.org/en/20/dialects/mysql.html

The resulting CREATE TABLE statement from the partitioning example, showing PARTITION BY, PARTITIONS, SUBPARTITION BY, and SUBPARTITIONS clauses.

```sql
CREATE TABLE testtable (
        id INTEGER NOT NULL AUTO_INCREMENT,
        other_id INTEGER NOT NULL,
        PRIMARY KEY (id, other_id)
)PARTITION BY KEY(other_id) PARTITIONS 2 SUBPARTITION BY HASH(some_expr) SUBPARTITIONS 2
```

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

### Inspect an AliasedClass with inspect() and aliased import

Source: https://docs.sqlalchemy.org/en/20/orm/queryguide/api.html

Use inspect() on an AliasedClass to get an AliasedInsp object. The example imports both inspect and aliased and assumes MyMappedClass is already mapped.

```python
from sqlalchemy import inspect
from sqlalchemy.orm import aliased

my_alias = aliased(MyMappedClass)
insp = inspect(my_alias)
```

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

### 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.
```

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

### 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
```

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

### 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()`

```

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

### Query.slice(start: int, stop: int) -> Self

Source: https://docs.sqlalchemy.org/en/20/orm/queryguide/query.html

Computes the “slice” of the Query represented by the given indices and returns the resulting Query. The start and stop indices behave like the argument to Python’s built-in range() function. This method provides an alternative to using LIMIT/OFFSET to get a slice of the query.

```APIDOC
## Query.slice(start: int, stop: int) -> Self

### Description
Computes the “slice” of the `Query` represented by the given indices and returns the resulting `Query`. The start and stop indices behave like the argument to Python’s built-in `range()` function. This method provides an alternative to using `LIMIT`/`OFFSET` to get a slice of the query.

### Method
`slice(start: int, stop: int) -> Self`

### Parameters
- **start** (int) - Required - The starting index.
- **stop** (int) - Required - The stopping index.

### Returns
- **Self** - A new Query object representing the slice.

### Example
```python
session.query(User).order_by(User.id).slice(1, 3)
```

Renders as:
```sql
SELECT users.id AS users_id,
       users.name AS users_name
FROM users ORDER BY users.id
LIMIT ? OFFSET ?
(2, 1)
```

### See Also
- `Query.limit()`
- `Query.offset()`
- `Select.slice()` - v2 equivalent method.
```

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

### Main function to insert and load sharded objects

Source: https://docs.sqlalchemy.org/en/20/_modules/examples/sharding/separate_tables.html

Calls setup, creates WeatherLocation and Report objects, and persists them using a Session. Demonstrates basic usage of the sharded setup.

```python
def main():
    setup()

    # save and load objects!

    tokyo = WeatherLocation("Asia", "Tokyo")
    newyork = WeatherLocation("North America", "New York")
    toronto = WeatherLocation("North America", "Toronto")
    london = WeatherLocation("Europe", "London")
    dublin = WeatherLocation("Europe", "Dublin")
    brasilia = WeatherLocation("South America", "Brasila")
    quito = WeatherLocation("South America", "Quito")

    tokyo.reports.append(Report(80.0))
    newyork.reports.append(Report(75))
    quito.reports.append(Report(85))

    with Session() as sess:
        sess.add_all(
            [tokyo, newyork, toronto, london, dublin, brasilia, quito]
        )

        sess.commit()
```
Tokens:
Raw