【Python百日进阶-Web开发-Peewee】Day284 - Peewee 的扩展（三）postgreSQL

文章目录

13.6.3 使用 hstore

首先，您需要从中导入自定义数据库类和 hstore 函数playhouse.postgres_ext（参见上面的代码片段）。然后，就像HStoreField在模型中添加 a 一样简单：

class House(BaseExtModel):
    address = CharField()
    features = HStoreField()

您现在可以在House实例上存储任意键/值对：

>>> h = House.create(
...     address='123 Main St',
...     features={'garage': '2 cars', 'bath': '2 bath'})
...
>>> h_from_db = House.get(House.id == h.id)
>>> h_from_db.features
{'bath': '2 bath', 'garage': '2 cars'}

您可以按单个键、多个键或部分字典进行过滤：

query = House.select()
garage = query.where(House.features.contains(‘garage’))
garage_and_bath = query.where(House.features.contains([‘garage’, ‘bath’]))
twocar = query.where(House.features.contains({‘garage’: ‘2 cars’}))
假设你想对房子进行原子更新：

>>> new_features = House.features.update({'bath': '2.5 bath', 'sqft': '1100'})
>>> query = House.update(features=new_features)
>>> query.where(House.id == h.id).execute()
1
>>> h = House.get(House.id == h.id)
>>> h.features
{'bath': '2.5 bath', 'garage': '2 cars', 'sqft': '1100'}

或者，或者原子删除：

>>> query = House.update(features=House.features.delete('bath'))
>>> query.where(House.id == h.id).execute()
1
>>> h = House.get(House.id == h.id)
>>> h.features
{'garage': '2 cars', 'sqft': '1100'}

可以同时删除多个键：

>>> query = House.update(features=House.features.delete('garage', 'sqft'))

您可以只选择键、值或压缩两者：

>>> for h in House.select(House.address, House.features.keys().alias('keys')):
...     print(h.address, h.keys)

123 Main St [u'bath', u'garage']

>>> for h in House.select(House.address, House.features.values().alias('vals')):
...     print(h.address, h.vals)

123 Main St [u'2 bath', u'2 cars']

>>> for h in House.select(House.address, House.features.items().alias('mtx')):
...     print(h.address, h.mtx)

123 Main St [[u'bath', u'2 bath'], [u'garage', u'2 cars']]

您可以检索数据切片，例如所有车库数据：

>>> query = House.select(House.address, House.features.slice('garage').alias('garage_data'))
>>> for house in query:
...     print(house.address, house.garage_data)

123 Main St {'garage': '2 cars'}

您可以检查是否存在键并相应地过滤行：

>>> has_garage = House.features.exists('garage')
>>> for house in House.select(House.address, has_garage.alias('has_garage')):
...     print(house.address, house.has_garage)

123 Main St True

>>> for house in House.select().where(House.features.exists('garage')):
...     print(house.address, house.features['garage'])  # <-- just houses w/garage data

123 Main St 2 cars

13.6.4 间隔支持 Interval support

Postgres 通过INTERVAL数据类型（docs）支持持续时间。

class IntervalField([null=False[, ...]])

能够存储 Pythondatetime.timedelta实例的字段类。

例子：

from datetime import timedelta

from playhouse.postgres_ext import *

db = PostgresqlExtDatabase('my_db')

class Event(Model):
    location = CharField()
    duration = IntervalField()
    start_time = DateTimeField()

    class Meta:
        database = db

    @classmethod
    def get_long_meetings(cls):
        return cls.select().where(cls.duration > timedelta(hours=1))

13.6.5 服务器端游标 Server-side cursors

当 psycopg2 执行查询时，通常所有结果都由后端获取并返回给客户端。这可能会导致您的应用程序在进行大型查询时使用大量内存。使用服务器端游标，一次返回一点结果（默认为 2000 条记录）。有关最终参考，请参阅psycopg2 文档。

笔记
要使用服务器端（或命名）游标，您必须使用PostgresqlExtDatabase.

要使用服务器端游标执行查询，只需使用ServerSide()帮助程序包装您的选择查询：

large_query = PageView.select()  # Build query normally.

# Iterate over large query inside a transaction.
for page_view in ServerSide(large_query):
    # do some interesting analysis here.
    pass

# Server-side resources are released.

如果您希望所有SELECT查询自动使用服务器端游标，您可以在创建时指定PostgresqlExtDatabase：

from postgres_ext import PostgresqlExtDatabase

ss_db = PostgresqlExtDatabase('my_db', server_side_cursors=True)

笔记
服务器端游标的生存时间与事务一样长，因此 peewee 不会commit()在执行SELECT 查询后自动调用。如果您commit在完成迭代后不这样做，您将不会释放服务器端资源，直到连接关闭（或事务稍后提交）。此外，由于 peewee 默认会缓存游标返回的行，因此您应该.iterator() 在迭代大型查询时始终调用。
如果您使用ServerSide()帮助程序，事务和调用iterator()将被透明地处理。

13.6.6 全文搜索 Full-text search

Postgresql使用特殊的数据类型（和）提供复杂的全文搜索。文档应存储或转换为类型，搜索查询应转换为 .tsvectortsquerytsvectortsquery

对于简单的情况，您可以简单地使用该Match()函数，它将自动执行适当的转换，并且不需要更改架构：

def blog_search(search_term):
    return Blog.select().where(
        (Blog.status == Blog.STATUS_PUBLISHED) &
        Match(Blog.content, search_term))

该Match()函数将自动将左侧操作数转换为 a tsvector，将右侧操作数转换为 a tsquery。为了获得更好的性能，建议您GIN在计划搜索的列上创建索引：

CREATE INDEX blog_full_text_search ON blog USING gin(to_tsvector(content));

或者，您可以使用TSVectorField来维护用于存储tsvector数据的专用列：

class Blog(Model):
    content = TextField()
    search_content = TSVectorField()

笔记
TSVectorField, 将自动使用 GIN 索引创建。

tsvector在插入或更新search_content字段时，您需要将传入的文本数据显式转换为：

content = 'Excellent blog post about peewee ORM.'
blog_entry = Blog.create(
    content=content,
    search_content=fn.to_tsvector(content))

要执行全文搜索，请使用TSVectorField.match()：

terms = 'python & (sqlite | postgres)'
results = Blog.select().where(Blog.search_content.match(terms))

有关详细信息，请参阅Postgres 全文搜索文档。

13.6.7 postgres_ext API 说明

class PostgresqlExtDatabase(database[, server_side_cursors=False[, register_hstore=False[, …]]])
与支持相同PostgresqlDatabase但需要：

参数：

database ( str ) – 要连接的数据库的名称。
server_side_cursors ( bool ) --SELECT查询是否应该使用服务器端游标。
register_hstore ( bool ) – 向连接注册 HStore 扩展。
服务器端游标
ArrayField
DateTimeTZField
JSONField
BinaryJSONField
HStoreField
TSVectorField
如果您希望使用 HStore 扩展，则必须指定register_hstore=True.

如果使用server_side_cursors，还请务必使用包装您的查询 ServerSide()。

ServerSide

ServerSide(select_query)

参数：选择查询– 一个SelectQuery实例。
Rtype 生成器：

将给定的选择查询包装在事务中，并调用其 iterator()方法以避免缓存行实例。为了释放服务器端资源，请务必耗尽生成器（遍历所有行）。

用法：

large_query = PageView.select()
for page_view in ServerSide(large_query):
    # Do something interesting.
    pass

# At this point server side resources are released.

class ArrayField

class ArrayField([field_class=IntegerField[, field_kwargs=None[, dimensions=1[, convert_values=False]]]])

参数：

field_class – 的子类Field，例如IntegerField。
field_kwargs ( dict ) – 要初始化的参数field_class。
dimensions ( int ) – 数组的维度。
convert_values ( bool ) – 将field_class值转换应用于数组数据。
能够存储提供的field_class数组的字段。

笔记
默认情况下 ArrayField 将使用 GIN 索引。要禁用此功能，请使用初始化字段index=False。

您可以存储和检索列表（或列表列表）：

class BlogPost(BaseModel):
    content = TextField()
    tags = ArrayField(CharField)


post = BlogPost(content='awesome', tags=['foo', 'bar', 'baz'])

此外，您可以使用__getitem__API 来查询数据库中的值或切片：

# Get the first tag on a given blog post.
first_tag = (BlogPost
             .select(BlogPost.tags[0].alias('first_tag'))
             .where(BlogPost.id == 1)
             .dicts()
             .get())

# first_tag = {'first_tag': 'foo'}

获取切片值：

# Get the first two tags.
two_tags = (BlogPost
            .select(BlogPost.tags[:2].alias('two'))
            .dicts()
            .get())
# two_tags = {'two': ['foo', 'bar']}

contains（*items）

参数：项目– 必须在给定数组字段中的一项或多项。

# Get all blog posts that are tagged with both "python" and "django".
Blog.select().where(Blog.tags.contains('python', 'django'))

contains_any（*items）

参数：项目– 在给定的数组字段中搜索一个或多个项目。
Like contains(), except 将匹配数组包含任何给定项目的行。

# Get all blog posts that are tagged with "flask" and/or "django".
Blog.select().where(Blog.tags.contains_any('flask', 'django'))

class DateTimeTZField

class DateTimeTZField( *args , **kwargs )

DateTimeField的时区感知子类。

class HStoreField

class HStoreField( *args , **kwargs )

用于存储和检索任意键/值对的字段。有关使用的详细信息，请参阅hstore 支持。

注意
要使用它，HStoreField您需要确保 hstore扩展已注册到连接。为此，请实例化PostgresqlExtDatabasewith register_hstore=True。

笔记
默认情况下HStoreField将使用GiST索引。要禁用此功能，请使用初始化字段index=False。

keys()

返回给定行的键。

>>> for h in House.select(House.address, House.features.keys().alias('keys')):
...     print(h.address, h.keys)

123 Main St [u'bath', u'garage']

values()

返回给定行的值。

>>> for h in House.select(House.address, House.features.values().alias('vals')):
...     print(h.address, h.vals)

123 Main St [u'2 bath', u'2 cars']

items()

像 python 一样dict，返回列表列表中的键和值：

>>> for h in House.select(House.address, House.features.items().alias('mtx')):
...     print(h.address, h.mtx)

123 Main St [[u'bath', u'2 bath'], [u'garage', u'2 cars']]

slice( *args )

返回给定键列表的数据片段。

>>> for h in House.select(House.address, House.features.slice('garage').alias('garage_data')):
...     print(h.address, h.garage_data)

123 Main St {'garage': '2 cars'}

exists(key)

查询给定键是否存在。

>>> for h in House.select(House.address, House.features.exists('garage').alias('has_garage')):
...     print(h.address, h.has_garage)

123 Main St True

>>> for h in House.select().where(House.features.exists('garage')):
...     print(h.address, h.features['garage']) # <-- just houses w/garage data

123 Main St 2 cars

defined(key)

查询给定键是否有与之关联的值。

update( **data)

对给定行的键/值执行原子更新。

>>> query = House.update(features=House.features.update(
...     sqft=2000,
...     year_built=2012))
>>> query.where(House.id == 1).execute()

delete（*keys）

删除给定行或行提供的键。

笔记
我们将使用UPDATE查询。

>>> query = House.update(features=House.features.delete(
...     'sqft', 'year_built'))
>>> query.where(House.id == 1).execute()

contains（value）

参数： value– a dict、 a listof 键或单个键。
查询行是否存在：

部分字典。
键列表。
一个键。

>>> query = House.select()
>>> has_garage = query.where(House.features.contains('garage'))
>>> garage_bath = query.where(House.features.contains(['garage', 'bath']))
>>> twocar = query.where(House.features.contains({'garage': '2 cars'}))

contains_any（*keys）

参数：钥匙– 一个或多个要搜索的键。
查询行是否存在任何键。

class JSONField

class JSONField（dumps=None，*args，**kwargs ）

参数：转储– 默认是调用 json.dumps() 或 dumps 函数。您可以覆盖此方法以创建自定义的 JSON 包装器。
适合存储和查询任意 JSON 的字段类。在模型上使用它时，将字段的值设置为 Python 对象（ a dict或 a list）。当您从数据库中检索值时，它将作为 Python 数据结构返回。

笔记
您必须使用 Postgres 9.2 / psycopg2 2.5 或更高版本。

笔记
如果您使用的是 Postgres 9.4，强烈考虑 BinaryJSONField改用它，因为它提供了更好的性能和更强大的查询选项。

示例模型声明：

db = PostgresqlExtDatabase('my_db')

class APIResponse(Model):
    url = CharField()
    response = JSONField()

    class Meta:
        database = db

存储 JSON 数据的示例：

url = 'http://foo.com/api/resource/'
resp = json.loads(urllib2.urlopen(url).read())
APIResponse.create(url=url, response=resp)

APIResponse.create(url='http://foo.com/baz/', response={'key': 'value'})

要查询，请使用 Python 的[]运算符来指定嵌套键或数组查找：

APIResponse.select().where(
    APIResponse.response['key1']['nested-key'] == 'some-value')

为了说明[]运算符的使用，假设我们将以下数据存储在中APIResponse：

{
  "foo": {
    "bar": ["i1", "i2", "i3"],
    "baz": {
      "huey": "mickey",
      "peewee": "nugget"
    }
  }
}

以下是一些查询的结果：

def get_data(expression):
    # Helper function to just retrieve the results of a
    # particular expression.
    query = (APIResponse
             .select(expression.alias('my_data'))
             .dicts()
             .get())
    return query['my_data']

# Accessing the foo -> bar subkey will return a JSON
# representation of the list.
get_data(APIResponse.data['foo']['bar'])
# '["i1", "i2", "i3"]'

# In order to retrieve this list as a Python list,
# we will call .as_json() on the expression.
get_data(APIResponse.data['foo']['bar'].as_json())
# ['i1', 'i2', 'i3']

# Similarly, accessing the foo -> baz subkey will
# return a JSON representation of the dictionary.
get_data(APIResponse.data['foo']['baz'])
# '{"huey": "mickey", "peewee": "nugget"}'

# Again, calling .as_json() will return an actual
# python dictionary.
get_data(APIResponse.data['foo']['baz'].as_json())
# {'huey': 'mickey', 'peewee': 'nugget'}

# When dealing with simple values, either way works as
# you expect.
get_data(APIResponse.data['foo']['bar'][0])
# 'i1'

# Calling .as_json() when the result is a simple value
# will return the same thing as the previous example.
get_data(APIResponse.data['foo']['bar'][0].as_json())
# 'i1'

class BinaryJSONField

class BinaryJSONField(dumps=None, *args, **kwargs)

参数：转储– 默认是调用 json.dumps() 或 dumps 函数。您可以覆盖此方法以创建自定义的 JSON 包装器。
存储和查询任意 JSON 文档。数据应该使用普通的 Pythondict和list对象存储，当从数据库返回数据时，也将使用dictand返回list。

有关基本查询操作的示例，请参见上述代码示例 JSONField。下面的示例查询将使用上述相同的 APIResponse模型。

笔记
默认情况下 BinaryJSONField 将使用 GiST 索引。要禁用此功能，请使用初始化字段index=False。

笔记
您必须使用 Postgres 9.4 / psycopg2 2.5 或更高版本。如果您使用的是 Postgres 9.2 或 9.3，则可以改用常规JSONField。

contains（other）

测试给定的 JSON 数据是否包含给定的 JSON 片段或键。

例子：

search_fragment = {
    'foo': {'bar': ['i2']}
}
query = (APIResponse
         .select()
         .where(APIResponse.data.contains(search_fragment)))

# If we're searching for a list, the list items do not need to
# be ordered in a particular way:
query = (APIResponse
         .select()
         .where(APIResponse.data.contains({
             'foo': {'bar': ['i2', 'i1']}})))

我们也可以传入简单的键。要查找包含foo顶级密钥的 APIResponses：

APIResponse.select().where(APIResponse.data.contains('foo'))

我们还可以使用方括号搜索子键：

APIResponse.select().where(
    APIResponse.data['foo']['bar'].contains(['i2', 'i1']))

contains_any（*items）

搜索一个或多个给定项目的存在。

APIResponse.select().where(
    APIResponse.data.contains_any('foo', 'baz', 'nugget'))

像contains()，我们也可以搜索子键：

APIResponse.select().where(
    APIResponse.data['foo']['bar'].contains_any('i2', 'ix'))

contains_all（*items）

搜索所有给定项目的存在。

APIResponse.select().where(
    APIResponse.data.contains_all('foo'))

像contains_any()，我们也可以搜索子键：

APIResponse.select().where(
    APIResponse.data['foo']['bar'].contains_all('i1', 'i2', 'i3'))

contained_by（other）

测试给定的 JSON 文档是否包含在给定的 JSON 文档中（是其子集）。该方法是的逆方法contains()。

big_doc = {
    'foo': {
        'bar': ['i1', 'i2', 'i3'],
        'baz': {
            'huey': 'mickey',
            'peewee': 'nugget',
        }
    },
    'other_key': ['nugget', 'bear', 'kitten'],
}
APIResponse.select().where(
    APIResponse.data.contained_by(big_doc))

concat(data）

连接两个字段数据和提供的数据。请注意，此操作不会合并或执行“深度连接”。

has_key(key)

测试该键是否存在于 JSON 对象的顶层。

remove（*keys）

从 JSON 对象的顶层移除一个或多个键。

Match

Match(field, query)

生成全文搜索表达式，自动将左侧操作数转换为 a tsvector，将右侧操作数自动转换为 a tsquery。

例子：

def blog_search(search_term):
    return Blog.select().where(
        (Blog.status == Blog.STATUS_PUBLISHED) &
        Match(Blog.content, search_term))

class TSVectorField

class TSVectorField

适合存储tsvector数据的字段类型。该字段将自动创建一个GIN索引以提高搜索性能。

笔记
存储在此字段中的数据仍需要手动转换为tsvector类型。

笔记
默认情况下，TSVectorField 将使用 GIN 索引。要禁用此功能，请使用初始化字段index=False。

示例用法：

class Blog(Model):
    content = TextField()
    search_content = TSVectorField()

content = 'this is a sample blog entry.'
blog_entry = Blog.create(
    content=content,
    search_content=fn.to_tsvector(content))  # Note `to_tsvector()`.

match(query[, query=None[, plain=False]])

参数：

query ( str ) – 全文搜索查询。
query( str ) – 语言名称（可选）。
plain ( bool ) – 使用普通（简单）解析器解析搜索查询。
返回：
表示全文搜索/匹配的表达式。

例子：

# Perform a search using the "match" method.
terms = 'python & (sqlite | postgres)'
results = Blog.select().where(Blog.search_content.match(terms))