自定义 postgres 范围类型的 Sqlalchemy 表示

Question

我需要自定义范围类型。我试图代表一个小时范围。对于一周中的每一天，一个范围（datetime.time，datetime.time）而不是单独的 TIME 列，如果可能的话，我希望能够访问 Postgres/sqlalchemy 范围运算符。

我正在寻找类似 TSRANGE 但小时而不是正常的时间 (datetime.datetime, datetime.datetime)

在 postgres 本身中，这非常有效。例如。

create type timerange as range (subtype = time);

create table schedule 
(
  id integer not null primary key,
  time_range timerange
);

insert into schedule 
values
(1, timerange(time '08:00', time '10:00', '[]')),
(2, timerange(time '10:00', time '12:00', '[]'));

select *
from schedule
where time_range @> time '09:00'

所以问题来了。如何在 SQLAlchemy 中表示我在 Postgres 中创建的这种自定义类型？子类 TSRANGE、TIME 上的 TypeDecorator，或者可能创建一个新的 SQLALchemy UserDefinedType。我不太确定该走哪条路。任何建议将不胜感激。谢谢！

Answer 1

为了使用自定义范围类型你need to dig a bit deeper:

在实例化使用这些列类型的模型时，您应该传递用于列类型的 DBAPI 驱动程序所期望的任何数据类型。对于psycopg2，这些是NumericRange、DateRange、DateTimeRange 和DateTimeTZRange或您在register_range() 注册的课程。

换句话说，您必须使用 DBAPI 注册您的自定义范围类型（​​通常是psycopg2）并创建 SQLAlchemy 类型以匹配注册的类型。 register_range() 采用 PostgreSQL range 类型的名称，Range 的（严格）子类和用于获取 oid 的连接/游标。它可以全局注册新的范围类型，也可以本地注册到给定的连接或游标：

In [2]: import psycopg2.extras

创建模型实例时应使用的值类型：

In [3]: class TimeRange(psycopg2.extras.Range):
   ...:     pass
   ...: 

在 SQLAlchemy 中使用 raw_connection() 获取底层 psycopg2 连接的代理。注册可能应该在实际实现中的设置函数中完成：

In [4]: conn = engine.raw_connection()

In [5]: cur = conn.cursor()

In [6]: psycopg2.extras.register_range('timerange', TimeRange, cur, globally=True)
Out[6]: <psycopg2._range.RangeCaster at 0x7f1c980dbe80>

In [7]: cur.close()

In [8]: conn.close()

接下来创建 SQLAlchemy 范围类型以匹配注册的TimeRange。 TypeDecorator 不适合，因为您没有使用现有类型。 UserDefinedType 应该是所有全新类型的基础。对于范围运算符，包括 RangeOperators mixin：

它被 postgres 方言中提供的所有范围类型使用，并且可能用于您自己创建的任何范围类型。

其余的几乎都是直接从the predefined range types复制过来的：

In [11]: from sqlalchemy.dialects import postgresql

In [13]: from sqlalchemy import types as sqltypes

In [14]: class TIMERANGE(postgresql.ranges.RangeOperators, sqltypes.UserDefinedType):
    ...:     def get_col_spec(self, **kw):
    ...:         return 'timerange'

这只是反射所必需的。

In [16]: postgresql.base.ischema_names['timerange'] = TIMERANGE

然后创建您的表格并照常使用：

In [17]: schedule = Table('schedule', metadata, autoload=True, autoload_with=engine)

In [18]: schedule
Out[18]: Table('schedule', MetaData(bind=Engine(postgresql:///sopython)), Column('id', INTEGER(), table=<schedule>, primary_key=True, nullable=False), Column('time_range', TIMERANGE(), table=<schedule>), schema=None)

In [19]: session.query(schedule).all()
Out[19]: 
[(1, TimeRange(datetime.time(8, 0), datetime.time(10, 0), '[]')),
 (2, TimeRange(datetime.time(10, 0), datetime.time(12, 0), '[]'))]

In [20]: session.query(schedule).\
    ...:     filter(schedule.c.time_range.contains(time(9, 0))).\
    ...:     all()
2017-04-11 10:01:23,864 INFO sqlalchemy.engine.base.Engine SELECT schedule.id AS schedule_id, schedule.time_range AS schedule_time_range 
FROM schedule 
WHERE schedule.time_range @> %(time_range_1)s
2017-04-11 10:01:23,864 INFO sqlalchemy.engine.base.Engine {'time_range_1': datetime.time(9, 0)}
Out[20]: [(1, TimeRange(datetime.time(8, 0), datetime.time(10, 0), '[]'))]