SqlAlchemy 2.0 中文文档（四十八）（1）

Table(
    "sometable",
    metadata,
    Column(
        "id", Integer, Sequence("some_id_seq", start=1), primary_key=True
    )
)

from sqlalchemy import Table, Column, MetaData, Integer, Computed
metadata = MetaData()
data = Table(
    "data",
    metadata,
    Column(
        'id', Integer, Identity(start=42, cycle=True), primary_key=True
    ),
    Column('data', String)
)

CREATE  TABLE  data  (
  id  INTEGER  GENERATED  BY  DEFAULT  AS  IDENTITY  (START  WITH  42  CYCLE),
  data  VARCHAR,
  PRIMARY  KEY  (id)
)

from sqlalchemy.schema import CreateColumn
from sqlalchemy.ext.compiler import compiles
@compiles(CreateColumn, 'postgresql')
def use_identity(element, compiler, **kw):
    text = compiler.visit_create_column(element, **kw)
    text = text.replace(
        "SERIAL", "INT GENERATED BY DEFAULT AS IDENTITY"
     )
    return text

t = Table(
    't', m,
    Column('id', Integer, primary_key=True),
    Column('data', String)
)

CREATE TABLE t (
    id INT GENERATED BY DEFAULT AS IDENTITY,
    data VARCHAR,
    PRIMARY KEY (id)
)
```## 服务器端游标
psycopg2、asyncpg 方言支持服务器端游标支持，其他方言也可能支持。
通过使用`Connection.execution_options.stream_results`连接执行选项，可以在每个语句基础上启用服务器端游标：
```py
with engine.connect() as conn:
    result = conn.execution_options(stream_results=True).execute(text("select * from table"))

engine = create_engine(
    "postgresql+pg8000://scott:tiger@localhost/test",
    isolation_level = "REPEATABLE READ"
)

with engine.connect() as conn:
    conn = conn.execution_options(
        isolation_level="REPEATABLE READ"
    )
    with conn.begin():
        # ... work with transaction

with engine.connect() as conn:
    conn = conn.execution_options(
        isolation_level="SERIALIZABLE",
        postgresql_readonly=True,
        postgresql_deferrable=True
    )
    with conn.begin():
        #  ... work with transaction

from sqlalchemy import create_engine
from sqlalchemy import event
postgresql_engine = create_engine(
    "postgresql+pyscopg2://scott:tiger@hostname/dbname",
    # disable default reset-on-return scheme
    pool_reset_on_return=None,
)
@event.listens_for(postgresql_engine, "reset")
def _reset_postgresql(dbapi_connection, connection_record, reset_state):
    if not reset_state.terminate_only:
        dbapi_connection.execute("CLOSE ALL")
        dbapi_connection.execute("RESET ALL")
        dbapi_connection.execute("DISCARD TEMP")
    # so that the DBAPI itself knows that the connection has been
    # reset
    dbapi_connection.rollback()

from sqlalchemy import event
from sqlalchemy import create_engine
engine = create_engine("postgresql+psycopg2://scott:tiger@host/dbname")
@event.listens_for(engine, "connect", insert=True)
def set_search_path(dbapi_connection, connection_record):
    existing_autocommit = dbapi_connection.autocommit
    dbapi_connection.autocommit = True
    cursor = dbapi_connection.cursor()
    cursor.execute("SET SESSION search_path='%s'" % schema_name)
    cursor.close()
    dbapi_connection.autocommit = existing_autocommit

test=> select current_schema();
current_schema
----------------
public
(1 row)

test=> select current_schema();
current_schema
----------------
public
(1 row)
test=> create schema scott;
CREATE SCHEMA
test=> select current_schema();
current_schema
----------------
scott
(1 row)

test=> show search_path;
search_path
-----------------
"$user", public
(1 row)

test=> CREATE TABLE test_schema.referred(id INTEGER PRIMARY KEY);
CREATE TABLE
test=> CREATE TABLE referring(
test(>         id INTEGER PRIMARY KEY,
test(>         referred_id INTEGER REFERENCES test_schema.referred(id));
CREATE TABLE
test=> SET search_path TO public, test_schema;
test=> SELECT pg_catalog.pg_get_constraintdef(r.oid, true) FROM
test-> pg_catalog.pg_class c JOIN pg_catalog.pg_namespace n
test-> ON n.oid = c.relnamespace
test-> JOIN pg_catalog.pg_constraint r  ON c.oid = r.conrelid
test-> WHERE c.relname='referring' AND r.contype = 'f'
test-> ;
               pg_get_constraintdef
---------------------------------------------------
 FOREIGN KEY (referred_id) REFERENCES referred(id)
(1 row)

test=> SET search_path TO public;
SET

test=> SELECT pg_catalog.pg_get_constraintdef(r.oid, true) FROM
test-> pg_catalog.pg_class c JOIN pg_catalog.pg_namespace n
test-> ON n.oid = c.relnamespace
test-> JOIN pg_catalog.pg_constraint r  ON c.oid = r.conrelid
test-> WHERE c.relname='referring' AND r.contype = 'f';
                     pg_get_constraintdef
---------------------------------------------------------------
 FOREIGN KEY (referred_id) REFERENCES test_schema.referred(id)
(1 row)

>>> from sqlalchemy import Table, MetaData, create_engine, text
>>> engine = create_engine("postgresql+psycopg2://scott:tiger@localhost/test")
>>> with engine.connect() as conn:
...     conn.execute(text("SET search_path TO test_schema, public"))
...     metadata_obj = MetaData()
...     referring = Table('referring', metadata_obj,
...                       autoload_with=conn)
...
<sqlalchemy.engine.result.CursorResult object at 0x101612ed0>

>>> metadata_obj.tables['referred'].schema is None
True

>>> with engine.connect() as conn:
...     conn.execute(text("SET search_path TO test_schema, public"))
...     metadata_obj = MetaData()
...     referring = Table('referring', metadata_obj,
...                       autoload_with=conn,
...                       postgresql_ignore_search_path=True)
...
<sqlalchemy.engine.result.CursorResult object at 0x1016126d0>

>>> metadata_obj.tables['test_schema.referred'].schema
'test_schema'

# INSERT..RETURNING
result = table.insert().returning(table.c.col1, table.c.col2).\
    values(name='foo')
print(result.fetchall())
# UPDATE..RETURNING
result = table.update().returning(table.c.col1, table.c.col2).\
    where(table.c.name=='foo').values(name='bar')
print(result.fetchall())
# DELETE..RETURNING
result = table.delete().returning(table.c.col1, table.c.col2).\
    where(table.c.name=='foo')
print(result.fetchall())

>>> from sqlalchemy.dialects.postgresql import insert
>>> insert_stmt = insert(my_table).values(
...     id='some_existing_id',
...     data='inserted value')
>>> do_nothing_stmt = insert_stmt.on_conflict_do_nothing(
...     index_elements=['id']
... )
>>> print(do_nothing_stmt)
INSERT  INTO  my_table  (id,  data)  VALUES  (%(id)s,  %(data)s)
ON  CONFLICT  (id)  DO  NOTHING
>>> do_update_stmt = insert_stmt.on_conflict_do_update(
...     constraint='pk_my_table',
...     set_=dict(data='updated value')
... )
>>> print(do_update_stmt)
INSERT  INTO  my_table  (id,  data)  VALUES  (%(id)s,  %(data)s)
ON  CONFLICT  ON  CONSTRAINT  pk_my_table  DO  UPDATE  SET  data  =  %(param_1)s 

>>> do_update_stmt = insert_stmt.on_conflict_do_update(
...     index_elements=['id'],
...     set_=dict(data='updated value')
... )
>>> print(do_update_stmt)
INSERT  INTO  my_table  (id,  data)  VALUES  (%(id)s,  %(data)s)
ON  CONFLICT  (id)  DO  UPDATE  SET  data  =  %(param_1)s
>>> do_update_stmt = insert_stmt.on_conflict_do_update(
...     index_elements=[my_table.c.id],
...     set_=dict(data='updated value')
... )
>>> print(do_update_stmt)
INSERT  INTO  my_table  (id,  data)  VALUES  (%(id)s,  %(data)s)
ON  CONFLICT  (id)  DO  UPDATE  SET  data  =  %(param_1)s 

>>> stmt = insert(my_table).values(user_email='a@b.com', data='inserted data')
>>> stmt = stmt.on_conflict_do_update(
...     index_elements=[my_table.c.user_email],
...     index_where=my_table.c.user_email.like('%@gmail.com'),
...     set_=dict(data=stmt.excluded.data)
... )
>>> print(stmt)
INSERT  INTO  my_table  (data,  user_email)
VALUES  (%(data)s,  %(user_email)s)  ON  CONFLICT  (user_email)
WHERE  user_email  LIKE  %(user_email_1)s  DO  UPDATE  SET  data  =  excluded.data 

>>> do_update_stmt = insert_stmt.on_conflict_do_update(
...     constraint='my_table_idx_1',
...     set_=dict(data='updated value')
... )
>>> print(do_update_stmt)
INSERT  INTO  my_table  (id,  data)  VALUES  (%(id)s,  %(data)s)
ON  CONFLICT  ON  CONSTRAINT  my_table_idx_1  DO  UPDATE  SET  data  =  %(param_1)s
>>> do_update_stmt = insert_stmt.on_conflict_do_update(
...     constraint='my_table_pk',
...     set_=dict(data='updated value')
... )
>>> print(do_update_stmt)
INSERT  INTO  my_table  (id,  data)  VALUES  (%(id)s,  %(data)s)
ON  CONFLICT  ON  CONSTRAINT  my_table_pk  DO  UPDATE  SET  data  =  %(param_1)s 

>>> do_update_stmt = insert_stmt.on_conflict_do_update(
...     constraint=my_table.primary_key,
...     set_=dict(data='updated value')
... )
>>> print(do_update_stmt)
INSERT  INTO  my_table  (id,  data)  VALUES  (%(id)s,  %(data)s)
ON  CONFLICT  (id)  DO  UPDATE  SET  data  =  %(param_1)s 

>>> stmt = insert(my_table).values(id='some_id', data='inserted value')
>>> do_update_stmt = stmt.on_conflict_do_update(
...     index_elements=['id'],
...     set_=dict(data='updated value')
... )
>>> print(do_update_stmt)
INSERT  INTO  my_table  (id,  data)  VALUES  (%(id)s,  %(data)s)
ON  CONFLICT  (id)  DO  UPDATE  SET  data  =  %(param_1)s 

>>> stmt = insert(my_table).values(
...     id='some_id',
...     data='inserted value',
...     author='jlh'
... )
>>> do_update_stmt = stmt.on_conflict_do_update(
...     index_elements=['id'],
...     set_=dict(data='updated value', author=stmt.excluded.author)
... )
>>> print(do_update_stmt)
INSERT  INTO  my_table  (id,  data,  author)
VALUES  (%(id)s,  %(data)s,  %(author)s)
ON  CONFLICT  (id)  DO  UPDATE  SET  data  =  %(param_1)s,  author  =  excluded.author 

>>> stmt = insert(my_table).values(
...     id='some_id',
...     data='inserted value',
...     author='jlh'
... )
>>> on_update_stmt = stmt.on_conflict_do_update(
...     index_elements=['id'],
...     set_=dict(data='updated value', author=stmt.excluded.author),
...     where=(my_table.c.status == 2)
... )
>>> print(on_update_stmt)
INSERT  INTO  my_table  (id,  data,  author)
VALUES  (%(id)s,  %(data)s,  %(author)s)
ON  CONFLICT  (id)  DO  UPDATE  SET  data  =  %(param_1)s,  author  =  excluded.author
WHERE  my_table.status  =  %(status_1)s 

>>> stmt = insert(my_table).values(id='some_id', data='inserted value')
>>> stmt = stmt.on_conflict_do_nothing(index_elements=['id'])
>>> print(stmt)
INSERT  INTO  my_table  (id,  data)  VALUES  (%(id)s,  %(data)s)
ON  CONFLICT  (id)  DO  NOTHING 

>>> stmt = insert(my_table).values(id='some_id', data='inserted value')
>>> stmt = stmt.on_conflict_do_nothing()
>>> print(stmt)
INSERT  INTO  my_table  (id,  data)  VALUES  (%(id)s,  %(data)s)
ON  CONFLICT  DO  NOTHING 
```## 全文搜索
PostgreSQL 的全文搜索系统可通过`func`命名空间来使用，结合使用自定义运算符通过`Operators.bool_op()`方法。对于一些简单情况并具有一定的跨后端兼容性，也可以使用`Operators.match()`运算符。
### 使用`match()`进行简单纯文本匹配
`Operators.match()`运算符提供跨兼容的简单文本匹配。对于 PostgreSQL 后端，它被硬编码为使用`@@`运算符与`plainto_tsquery()` PostgreSQL 函数结合生成表达式。
在 PostgreSQL 方言中，类似以下表达式：
```py
select(sometable.c.text.match("search string"))

SELECT text @@ plainto_tsquery('search string') FROM table

from sqlalchemy import func
select(
    sometable.c.text.bool_op("@@")(func.to_tsquery("search string"))
)

SELECT text @@ to_tsquery('search string') FROM table

select(
    func.to_tsquery('cat').bool_op("@>")(func.to_tsquery('cat & rat'))
)

SELECT  to_tsquery('cat')  @>  to_tsquery('cat & rat')

from sqlalchemy.dialects.postgresql import TSVECTOR
from sqlalchemy import select, cast
select(cast("some text", TSVECTOR))

SELECT CAST('some text' AS TSVECTOR) AS anon_1

select(mytable.c.id).where(
    mytable.c.title.match('somestring', postgresql_regconfig='english')
)

SELECT mytable.id FROM mytable
WHERE mytable.title @@ plainto_tsquery('english', 'somestring')

select(mytable.c.id).where(
    func.to_tsvector("english", mytable.c.title).bool_op("@@")(
        func.to_tsquery("english", "somestring")
    )
)

SELECT mytable.id FROM mytable
WHERE to_tsvector('english', mytable.title) @@
    to_tsquery('english', 'somestring')

# SELECT ... FROM ONLY ...
result = table.select().with_hint(table, 'ONLY', 'postgresql')
print(result.fetchall())
# UPDATE ONLY ...
table.update(values=dict(foo='bar')).with_hint('ONLY',
                                               dialect_name='postgresql')
# DELETE FROM ONLY ...
table.delete().with_hint('ONLY', dialect_name='postgresql')

Index("my_index", table.c.x, postgresql_include=['y'])

Index('my_index', my_table.c.id, postgresql_where=my_table.c.value > 10)
```### 操作符类
PostgreSQL 允许为索引的每一列指定*操作符类*（参见[`www.postgresql.org/docs/current/interactive/indexes-opclass.html`](https://www.postgresql.org/docs/current/interactive/indexes-opclass.html)）。`Index`构造允许通过`postgresql_ops`关键字参数指定这些：
```py
Index(
    'my_index', my_table.c.id, my_table.c.data,
    postgresql_ops={
        'data': 'text_pattern_ops',
        'id': 'int4_ops'
    })

Index(
    'my_index', my_table.c.id,
    func.lower(my_table.c.data).label('data_lower'),
    postgresql_ops={
        'data_lower': 'text_pattern_ops',
        'id': 'int4_ops'
    })

Index('my_index', my_table.c.data, postgresql_using='gin')

Index('my_index', my_table.c.data, postgresql_with={"fillfactor": 50})

Index('my_index', my_table.c.data, postgresql_tablespace='my_tablespace')

tbl = Table('testtbl', m, Column('data', Integer))
idx1 = Index('test_idx1', tbl.c.data, postgresql_concurrently=True)

CREATE INDEX CONCURRENTLY test_idx1 ON testtbl (data)

DROP INDEX CONCURRENTLY test_idx1

metadata = MetaData()
table = Table(
    "foo", metadata,
    Column("id", String))
index = Index(
    "foo_idx", table.c.id, postgresql_concurrently=True)
with engine.connect() as conn:
    with conn.execution_options(isolation_level='AUTOCOMMIT'):
        table.create(conn)

from sqlalchemy import create_engine, inspect
engine = create_engine("postgresql+psycopg2://localhost/test")
insp = inspect(engine)  # will be a PGInspector
print(insp.get_enums())

class sqlalchemy.dialects.postgresql.base.PGInspector

method get_domains(schema: str | None = None) → List[ReflectedDomain]

method get_enums(schema: str | None = None) → List[ReflectedEnum]

method get_foreign_table_names(schema: str | None = None) → List[str]

method get_table_oid(table_name: str, schema: str | None = None) → int

method has_type(type_name: str, schema: str | None = None, **kw: Any) → bool

Table("some_table", metadata, ..., postgresql_inherits="some_supertable")
Table("some_table", metadata, ..., postgresql_inherits=("t1", "t2", ...))

Table("some_table", metadata, ..., postgresql_on_commit='PRESERVE ROWS')

Table("some_table", metadata, ...,
      postgresql_partition_by='LIST (part_column)')
.. versionadded:: 1.2.6

Table("some_table", metadata, ..., postgresql_tablespace='some_tablespace')

Table("some_table", metadata, ..., postgresql_using='heap')
.. versionadded:: 2.0.26

Table("some_table", metadata, ..., postgresql_with_oids=True)

Table("some_table", metadata, ..., postgresql_with_oids=False)

def update():
    op.create_foreign_key(
        "fk_user_address",
        "address",
        "user",
        ["user_id"],
        ["id"],
        postgresql_not_valid=True
    )

CheckConstraint("some_field IS NOT NULL", postgresql_not_valid=True)
ForeignKeyConstraint(["some_id"], ["some_table.some_id"], postgresql_not_valid=True)

>>> from sqlalchemy import select, func
>>> stmt = select(func.json_each('{"a":"foo", "b":"bar"}').table_valued("key", "value"))
>>> print(stmt)
SELECT  anon_1.key,  anon_1.value
FROM  json_each(:json_each_1)  AS  anon_1 

>>> from sqlalchemy import select, func, literal_column
>>> stmt = select(
...     func.json_populate_record(
...         literal_column("null::myrowtype"),
...         '{"a":1,"b":2}'
...     ).table_valued("a", "b", name="x")
... )
>>> print(stmt)
SELECT  x.a,  x.b
FROM  json_populate_record(null::myrowtype,  :json_populate_record_1)  AS  x

>>> from sqlalchemy import select, func, column, Integer, Text
>>> stmt = select(
...     func.json_to_record('{"a":1,"b":[1,2,3],"c":"bar"}').table_valued(
...         column("a", Integer), column("b", Text), column("d", Text),
...     ).render_derived(name="x", with_types=True)
... )
>>> print(stmt)
SELECT  x.a,  x.b,  x.d
FROM  json_to_record(:json_to_record_1)  AS  x(a  INTEGER,  b  TEXT,  d  TEXT) 

>>> from sqlalchemy import select, func
>>> stmt = select(
...     func.generate_series(4, 1, -1).
...     table_valued("value", with_ordinality="ordinality").
...     render_derived()
... )
>>> print(stmt)
SELECT  anon_1.value,  anon_1.ordinality
FROM  generate_series(:generate_series_1,  :generate_series_2,  :generate_series_3)
WITH  ORDINALITY  AS  anon_1(value,  ordinality) 

>>> from sqlalchemy import select, func
>>> stmt = select(func.json_array_elements('["one", "two"]').column_valued("x"))
>>> print(stmt)
SELECT  x
FROM  json_array_elements(:json_array_elements_1)  AS  x 

>>> from sqlalchemy.dialects.postgresql import array
>>> from sqlalchemy import select, func
>>> stmt = select(func.unnest(array([1, 2])).column_valued())
>>> print(stmt)
SELECT  anon_1
FROM  unnest(ARRAY[%(param_1)s,  %(param_2)s])  AS  anon_1 

>>> from sqlalchemy import table, column, ARRAY, Integer
>>> from sqlalchemy import select, func
>>> t = table("t", column('value', ARRAY(Integer)))
>>> stmt = select(func.unnest(t.c.value).column_valued("unnested_value"))
>>> print(stmt)
SELECT  unnested_value
FROM  unnest(t.value)  AS  unnested_value 

>>> from sqlalchemy import table, column, func, tuple_
>>> t = table("t", column("id"), column("fk"))
>>> stmt = t.select().where(
...     tuple_(t.c.id, t.c.fk) > (1,2)
... ).where(
...     func.ROW(t.c.id, t.c.fk) < func.ROW(3, 7)
... )
>>> print(stmt)
SELECT  t.id,  t.fk
FROM  t
WHERE  (t.id,  t.fk)  >  (:param_1,  :param_2)  AND  ROW(t.id,  t.fk)  <  ROW(:ROW_1,  :ROW_2) 

>>> from sqlalchemy import table, column, func, select
>>> a = table( "a", column("id"), column("x"), column("y"))
>>> stmt = select(func.row_to_json(a.table_valued()))
>>> print(stmt)
SELECT  row_to_json(a)  AS  row_to_json_1
FROM  a 

from sqlalchemy import TypeDecorator
from sqlalchemy.dialects.postgresql import ARRAY
class ArrayOfEnum(TypeDecorator):
    impl = ARRAY
    def bind_expression(self, bindvalue):
        return sa.cast(bindvalue, self)
    def result_processor(self, dialect, coltype):
        super_rp = super(ArrayOfEnum, self).result_processor(dialect, coltype)
        def handle_raw_string(value):
            inner = re.match(r"^{(.*)}$", value).group(1)
            return inner.split(",") if inner else []
        def process(value):
            if value is None:
                return None
            return super_rp(handle_raw_string(value))
        return process

Table(
    "mydata",
    metadata,
    Column("id", Integer, primary_key=True),
    Column("data", ArrayOfEnum(ENUM("a", "b", "c", name="myenum"))),
)

class CastingArray(ARRAY):
    def bind_expression(self, bindvalue):
        return sa.cast(bindvalue, self)

Table(
    "mydata",
    metadata,
    Column("id", Integer, primary_key=True),
    Column("data", CastingArray(JSONB)),
)
```## 范围和多范围类型
PostgreSQL 支持 psycopg、pg8000 和 asyncpg 方言的范围和多范围类型；psycopg2 方言仅支持范围类型。
新版本 2.0.17 中新增了对 pg8000 方言的范围和多范围支持。需要 pg8000 1.29.8 或更高版本。
传递给数据库的数据值可以作为字符串值传递，也可以使用 `Range` 数据对象。
新版本 2.0 中：添加了后端不可知的 `Range` 对象，用于指示范围。`psycopg2` 特定的范围类不再公开，仅在该特定方言内部使用。
例如，使用 `TSRANGE` 数据类型的完全类型化模型示例：
```py
from datetime import datetime
from sqlalchemy.dialects.postgresql import Range
from sqlalchemy.dialects.postgresql import TSRANGE
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column
class Base(DeclarativeBase):
    pass
class RoomBooking(Base):
    __tablename__ = "room_booking"
    id: Mapped[int] = mapped_column(primary_key=True)
    room: Mapped[str]
    during: Mapped[Range[datetime]] = mapped_column(TSRANGE)

from sqlalchemy import create_engine
from sqlalchemy.orm import Session
engine = create_engine("postgresql+psycopg://scott:tiger@pg14/dbname")
Base.metadata.create_all(engine)
with Session(engine) as session:
    booking = RoomBooking(
        room="101", during=Range(datetime(2013, 3, 23), datetime(2013, 3, 25))
    )
    session.add(booking)
    session.commit()

from sqlalchemy import select
with Session(engine) as session:
    for row in session.execute(select(RoomBooking.during)):
        print(row)

class sqlalchemy.dialects.postgresql.Range

r = Range(10, 50, bounds="()")

method __eq__(other: Any) → bool

method adjacent_to(other: Range[_T]) → bool

method contained_by(other: Range[_T]) → bool

method contains(value: _T | Range[_T]) → bool

method difference(other: Range[_T]) → Range[_T]

method intersection(other: Range[_T]) → Range[_T]

attribute is_empty

attribute isempty

attribute lower: _T | None

attribute lower_inc

attribute lower_inf

method not_extend_left_of(other: Range[_T]) → bool

method not_extend_right_of(other: Range[_T]) → bool

method overlaps(other: Range[_T]) → bool

method strictly_left_of(other: Range[_T]) → bool

method strictly_right_of(other: Range[_T]) → bool

method union(other: Range[_T]) → Range[_T]

attribute upper: _T | None

attribute upper_inc

attribute upper_inf

from datetime import datetime
from typing import List
from sqlalchemy.dialects.postgresql import Range
from sqlalchemy.dialects.postgresql import TSMULTIRANGE
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column
class Base(DeclarativeBase):
    pass
class EventCalendar(Base):
    __tablename__ = "event_calendar"
    id: Mapped[int] = mapped_column(primary_key=True)
    event_name: Mapped[str]
    added: Mapped[datetime]
    in_session_periods: Mapped[List[Range[datetime]]] = mapped_column(TSMULTIRANGE)

from sqlalchemy import create_engine
from sqlalchemy import select
from sqlalchemy.orm import Session
engine = create_engine("postgresql+psycopg://scott:tiger@pg14/test")
Base.metadata.create_all(engine)
with Session(engine) as session:
    calendar = EventCalendar(
        event_name="SQLAlchemy Tutorial Sessions",
        in_session_periods=[
            Range(datetime(2013, 3, 23), datetime(2013, 3, 25)),
            Range(datetime(2013, 4, 12), datetime(2013, 4, 15)),
            Range(datetime(2013, 5, 9), datetime(2013, 5, 12)),
        ],
    )
    session.add(calendar)
    session.commit()
    for multirange in session.scalars(select(EventCalendar.in_session_periods)):
        for range_ in multirange:
            print(f"Start: {range_.lower} End: {range_.upper}")

from sqlalchemy import literal
from sqlalchemy.dialects.postgresql import MultiRange
with Session(engine) as session:
    stmt = select(EventCalendar).where(
        EventCalendar.added.op("<@")(
            MultiRange(
                [
                    Range(datetime(2023, 1, 1), datetime(2013, 3, 31)),
                    Range(datetime(2023, 7, 1), datetime(2013, 9, 30)),
                ]
            )
        )
    )
    in_range = session.execute(stmt).all()
with engine.connect() as conn:
    row = conn.scalar(select(literal(MultiRange([Range(2, 4)]))))
    print(f"{row.lower} -> {row.upper}")

e = create_engine(
    "postgresql+psycopg://scott:tiger@host/dbname", native_inet_types=False
)