自定义 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 中创建的这个自定义类型？在 TIME 上将 TSRANGE、TypeDecorator 子类化，或者可能创建一个新的 SQLALchemy UserDefinedType。我不太确定要走哪条路。任何建议将不胜感激。谢谢！

Answer 1

为了使用自定义范围类型，您需要更深入地挖掘：

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

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

In [2]: import psycopg2.extras

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

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

raw_connection()在 SQLAlchemy 中使用以获取到底层psycopg2连接的代理。在实际实现中，注册也许应该在 setup 函数中完成：

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. ATypeDecorator不适合，因为您没有使用现有类型。UserDefinedType应该是所有全新类型的基础。对于范围运算符，包括RangeOperatorsmixin：

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

其余的几乎直接从预定义的范围类型复制：

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), '[]'))]