python - 自定义 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

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

When instantiating models that use these column types, you should pass whatever data type is expected by the DBAPI driver you’re using for the column type. For psycopg2 these are NumericRange, DateRange, DateTimeRange and DateTimeTZRange or the class you’ve registered with register_range().

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

In [2]: import psycopg2.extras

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

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

使用raw_connection()在 SQLAlchemy 中获取底层 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混合:

It is used by all the range types provided in the postgres dialect and can likely be used for any range types you create yourself.

其余部分几乎是直接从 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), '[]'))]