@clawhub-ygp987-2e9d4dc4ce
通过瀚高数据库（HighGo DB）提供的自定义 psycopg2 驱动连接数据库。该驱动已内置于技能中，支持 Python 2.7 和国密 SM4 安全特性。
---
name: highgo-db
description: 通过瀚高数据库（HighGo DB）提供的自定义 psycopg2 驱动连接数据库。该驱动已内置于技能中，支持 Python 2.7 和国密 SM4 安全特性。
---
# 瀚高数据库 (HighGo DB) 连接技能
此技能允许使用瀚高提供的修改版 `psycopg2` 驱动连接瀚高数据库。该驱动已内置在技能的 `assets/driver/` 目录下。
**注意：内置驱动是针对 Python 2.7 编译的，必须在 Python 2.7 环境下运行。**
## 核心功能
1. **执行 SQL 语句**: 在瀚高数据库中运行查询或管理指令。
2. **内置驱动驱动支持**: 自动加载集成在技能内部的瀚高专用 `psycopg2` 驱动。
3. **环境适配**: 脚本已适配 Python 2.7，并能自动处理内部共享库（.so）的加载。
## 运行环境要求
- **Python**: 必须安装 Python 2.7。
- **系统库**:
- `libpython2.7.so.1.0`
- `libldap_r-2.4.so.2` (或兼容版本)
- **Arch Linux 用户**: 如遇到依赖或 PGP 密钥问题，请参阅 [references/driver_info.md](references/driver_info.md) 的安装建议。
## 使用指南
### 1. 执行查询
直接使用 `python2` 调用技能内部的 `execute_query.py`。由于驱动已内置，通常**无需**指定驱动路径。
**示例命令：**
```bash
python2 highgo-db/scripts/execute_query.py \
--dsn "host=10.238.18.128 port=5866 dbname=ficc user=fic password='PASSWORD' options='-c search_path=system'" \
--sql "SELECT count(*) FROM sys_user;"
```
### 2. 注意事项
- **DSN 参数**: 瀚高数据库通常需要指定 `options='-c search_path=system'` 来正确访问系统表。
- **驱动覆盖**: 如果需要使用外部驱动，仍可通过 `--driver` 参数手动指定。
## 参考资料
- 关于驱动的详细信息及 Arch Linux 安装建议，请参阅 [references/driver_info.md](references/driver_info.md)。
FILE:README.md
# HighGo DB Skill for Gemini CLI
这是一个为 Gemini CLI 开发的自包含技能，旨在通过瀚高数据库（HighGo DB）提供的自定义 `psycopg2` 驱动连接并操作数据库。
## 特性
- **内置驱动**: 集成了瀚高专用 `psycopg2` 驱动（支持国密 SM4 等安全特性）。
- **Python 2.7 适配**: 核心脚本已完全适配 Python 2.7 环境，解决了旧版驱动在 Python 3 下的兼容性问题。
- **自动环境配置**: 运行时自动配置 `LD_LIBRARY_PATH` 以正确加载驱动依赖的 `.so` 库。
- **SDD 流程支持**: 适用于规范驱动开发流程中的数据库验证和查询。
## 运行环境要求
由于瀚高提供的二进制驱动是针对 Python 2.7 编译的，因此环境必须满足：
- **Python**: 2.7.x
- **系统库**:
- `libpython2.7.so.1.0`
- `libldap_r-2.4.so.2` (或兼容版本)
- `openssl-1.1` (旧版 OpenSSL 支持)
### Arch Linux 安装建议
```bash
# 导入必要的 PGP 密钥
gpg --keyserver hkps://keyserver.ubuntu.com --recv-keys A21FAB74B0088AA361152586B8EF1A6BA9DA2D5C EFC0A467D613CB83C7ED6D30D894E2CE8B3D79F5 C01E1CAD5EA2C4F0B8E3571504C367C218ADD4FF 8657ABB260F056B1E5190839D9C4D26D0E604491 7953AC1FBC3DC8B3B292393ED5E9E43F7DF9EE8C
# 从 AUR 安装依赖
# 建议使用 yay 或 makepkg 手动安装 python2 和 openssl-1.1
```
## 安装技能
1. 下载或克隆此仓库。
2. 在项目根目录（或技能目录）执行安装：
```bash
gemini skills install highgo-db.skill --scope workspace
```
3. 在交互界面执行 `/skills reload`。
## 使用方法
### 执行查询
直接使用 `python2` 调用技能内部的脚本：
```bash
python2 highgo-db/scripts/execute_query.py \
--dsn "host=[IP] port=[PORT] dbname=[DB] user=[USER] password='[PASS]' options='-c search_path=system'" \
--sql "SELECT count(*) FROM sys_user;"
```
## 项目结构
- `SKILL.md`: 技能元数据与触发规则。
- `scripts/`: 包含 Python 2.7 兼容的查询执行脚本。
- `assets/driver/`: 内置的瀚高专用驱动文件。
- `references/`: 驱动详细信息与故障排除指南。
## 许可证
基于瀚高数据库提供的 `psycopg2` 修改版，遵循 LGPL 协议。
FILE:assets/driver/psycopg2/__init__.py
"""A Python driver for PostgreSQL
psycopg is a PostgreSQL_ database adapter for the Python_ programming
language. This is version 2, a complete rewrite of the original code to
provide new-style classes for connection and cursor objects and other sweet
candies. Like the original, psycopg 2 was written with the aim of being very
small and fast, and stable as a rock.
Homepage: https://psycopg.org/
.. _PostgreSQL: https://www.postgresql.org/
.. _Python: https://www.python.org/
:Groups:
* `Connections creation`: connect
* `Value objects constructors`: Binary, Date, DateFromTicks, Time,
TimeFromTicks, Timestamp, TimestampFromTicks
"""
# psycopg/__init__.py - initialization of the psycopg module
#
# Copyright (C) 2003-2019 Federico Di Gregorio <[email protected]>
# Copyright (C) 2020 The Psycopg Team
#
# psycopg2 is free software: you can redistribute it and/or modify it
# under the terms of the GNU Lesser General Public License as published
# by the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# In addition, as a special exception, the copyright holders give
# permission to link this program with the OpenSSL library (or with
# modified versions of OpenSSL that use the same license as OpenSSL),
# and distribute linked combinations including the two.
#
# You must obey the GNU Lesser General Public License in all respects for
# all of the code used other than OpenSSL.
#
# psycopg2 is distributed in the hope that it will be useful, but WITHOUT
# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
# FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public
# License for more details.
# Import modules needed by _psycopg to allow tools like py2exe to do
# their work without bothering about the module dependencies.
# Note: the first internal import should be _psycopg, otherwise the real cause
# of a failed loading of the C module may get hidden, see
# https://archives.postgresql.org/psycopg/2011-02/msg00044.php
# Import the DBAPI-2.0 stuff into top-level module.
from psycopg2._psycopg import ( # noqa
BINARY, NUMBER, STRING, DATETIME, ROWID,
Binary, Date, Time, Timestamp,
DateFromTicks, TimeFromTicks, TimestampFromTicks,
Error, Warning, DataError, DatabaseError, ProgrammingError, IntegrityError,
InterfaceError, InternalError, NotSupportedError, OperationalError,
_connect, apilevel, threadsafety, paramstyle,
__version__, __libpq_version__,
)
from psycopg2 import tz # noqa
# Register default adapters.
from psycopg2 import extensions as _ext
_ext.register_adapter(tuple, _ext.SQL_IN)
_ext.register_adapter(type(None), _ext.NoneAdapter)
# Register the Decimal adapter here instead of in the C layer.
# This way a new class is registered for each sub-interpreter.
# See ticket #52
from decimal import Decimal # noqa
from psycopg2._psycopg import Decimal as Adapter # noqa
_ext.register_adapter(Decimal, Adapter)
del Decimal, Adapter
def connect(dsn=None, connection_factory=None, cursor_factory=None, **kwargs):
"""
Create a new database connection.
The connection parameters can be specified as a string:
conn = psycopg2.connect("dbname=test user=postgres password=secret")
or using a set of keyword arguments:
conn = psycopg2.connect(database="test", user="postgres", password="secret")
Or as a mix of both. The basic connection parameters are:
- *dbname*: the database name
- *database*: the database name (only as keyword argument)
- *user*: user name used to authenticate
- *password*: password used to authenticate
- *host*: database host address (defaults to UNIX socket if not provided)
- *port*: connection port number (defaults to 5432 if not provided)
Using the *connection_factory* parameter a different class or connections
factory can be specified. It should be a callable object taking a dsn
argument.
Using the *cursor_factory* parameter, a new default cursor factory will be
used by cursor().
Using *async*=True an asynchronous connection will be created. *async_* is
a valid alias (for Python versions where ``async`` is a keyword).
Any other keyword parameter will be passed to the underlying client
library: the list of supported parameters depends on the library version.
"""
kwasync = {}
if 'async' in kwargs:
kwasync['async'] = kwargs.pop('async')
if 'async_' in kwargs:
kwasync['async_'] = kwargs.pop('async_')
if dsn is None and not kwargs:
raise TypeError('missing dsn and no parameters')
dsn = _ext.make_dsn(dsn, **kwargs)
conn = _connect(dsn, connection_factory=connection_factory, **kwasync)
if cursor_factory is not None:
conn.cursor_factory = cursor_factory
return conn
FILE:assets/driver/psycopg2/_ipaddress.py
"""Implementation of the ipaddres-based network types adaptation
"""
# psycopg/_ipaddress.py - Ipaddres-based network types adaptation
#
# Copyright (C) 2016-2019 Daniele Varrazzo <[email protected]>
# Copyright (C) 2020 The Psycopg Team
#
# psycopg2 is free software: you can redistribute it and/or modify it
# under the terms of the GNU Lesser General Public License as published
# by the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# In addition, as a special exception, the copyright holders give
# permission to link this program with the OpenSSL library (or with
# modified versions of OpenSSL that use the same license as OpenSSL),
# and distribute linked combinations including the two.
#
# You must obey the GNU Lesser General Public License in all respects for
# all of the code used other than OpenSSL.
#
# psycopg2 is distributed in the hope that it will be useful, but WITHOUT
# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
# FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public
# License for more details.
from psycopg2.extensions import (
new_type, new_array_type, register_type, register_adapter, QuotedString)
from psycopg2.compat import text_type
# The module is imported on register_ipaddress
ipaddress = None
# The typecasters are created only once
_casters = None
def register_ipaddress(conn_or_curs=None):
"""
Register conversion support between `ipaddress` objects and `network types`__.
:param conn_or_curs: the scope where to register the type casters.
If `!None` register them globally.
After the function is called, PostgreSQL :sql:`inet` values will be
converted into `~ipaddress.IPv4Interface` or `~ipaddress.IPv6Interface`
objects, :sql:`cidr` values into into `~ipaddress.IPv4Network` or
`~ipaddress.IPv6Network`.
.. __: https://www.postgresql.org/docs/current/static/datatype-net-types.html
"""
global ipaddress
import ipaddress
global _casters
if _casters is None:
_casters = _make_casters()
for c in _casters:
register_type(c, conn_or_curs)
for t in [ipaddress.IPv4Interface, ipaddress.IPv6Interface,
ipaddress.IPv4Network, ipaddress.IPv6Network]:
register_adapter(t, adapt_ipaddress)
def _make_casters():
inet = new_type((869,), 'INET', cast_interface)
ainet = new_array_type((1041,), 'INET[]', inet)
cidr = new_type((650,), 'CIDR', cast_network)
acidr = new_array_type((651,), 'CIDR[]', cidr)
return [inet, ainet, cidr, acidr]
def cast_interface(s, cur=None):
if s is None:
return None
# Py2 version force the use of unicode. meh.
return ipaddress.ip_interface(text_type(s))
def cast_network(s, cur=None):
if s is None:
return None
return ipaddress.ip_network(text_type(s))
def adapt_ipaddress(obj):
return QuotedString(str(obj))
FILE:assets/driver/psycopg2/_json.py
"""Implementation of the JSON adaptation objects
This module exists to avoid a circular import problem: pyscopg2.extras depends
on psycopg2.extension, so I can't create the default JSON typecasters in
extensions importing register_json from extras.
"""
# psycopg/_json.py - Implementation of the JSON adaptation objects
#
# Copyright (C) 2012-2019 Daniele Varrazzo <[email protected]>
# Copyright (C) 2020 The Psycopg Team
#
# psycopg2 is free software: you can redistribute it and/or modify it
# under the terms of the GNU Lesser General Public License as published
# by the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# In addition, as a special exception, the copyright holders give
# permission to link this program with the OpenSSL library (or with
# modified versions of OpenSSL that use the same license as OpenSSL),
# and distribute linked combinations including the two.
#
# You must obey the GNU Lesser General Public License in all respects for
# all of the code used other than OpenSSL.
#
# psycopg2 is distributed in the hope that it will be useful, but WITHOUT
# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
# FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public
# License for more details.
import json
from psycopg2._psycopg import ISQLQuote, QuotedString
from psycopg2._psycopg import new_type, new_array_type, register_type
from psycopg2.compat import PY2
# oids from PostgreSQL 9.2
JSON_OID = 114
JSONARRAY_OID = 199
# oids from PostgreSQL 9.4
JSONB_OID = 3802
JSONBARRAY_OID = 3807
class Json(object):
"""
An `~psycopg2.extensions.ISQLQuote` wrapper to adapt a Python object to
:sql:`json` data type.
`!Json` can be used to wrap any object supported by the provided *dumps*
function. If none is provided, the standard :py:func:`json.dumps()` is
used.
"""
def __init__(self, adapted, dumps=None):
self.adapted = adapted
self._conn = None
self._dumps = dumps or json.dumps
def __conform__(self, proto):
if proto is ISQLQuote:
return self
def dumps(self, obj):
"""Serialize *obj* in JSON format.
The default is to call `!json.dumps()` or the *dumps* function
provided in the constructor. You can override this method to create a
customized JSON wrapper.
"""
return self._dumps(obj)
def prepare(self, conn):
self._conn = conn
def getquoted(self):
s = self.dumps(self.adapted)
qs = QuotedString(s)
if self._conn is not None:
qs.prepare(self._conn)
return qs.getquoted()
if PY2:
def __str__(self):
return self.getquoted()
else:
def __str__(self):
# getquoted is binary in Py3
return self.getquoted().decode('ascii', 'replace')
def register_json(conn_or_curs=None, globally=False, loads=None,
oid=None, array_oid=None, name='json'):
"""Create and register typecasters converting :sql:`json` type to Python objects.
:param conn_or_curs: a connection or cursor used to find the :sql:`json`
and :sql:`json[]` oids; the typecasters are registered in a scope
limited to this object, unless *globally* is set to `!True`. It can be
`!None` if the oids are provided
:param globally: if `!False` register the typecasters only on
*conn_or_curs*, otherwise register them globally
:param loads: the function used to parse the data into a Python object. If
`!None` use `!json.loads()`, where `!json` is the module chosen
according to the Python version (see above)
:param oid: the OID of the :sql:`json` type if known; If not, it will be
queried on *conn_or_curs*
:param array_oid: the OID of the :sql:`json[]` array type if known;
if not, it will be queried on *conn_or_curs*
:param name: the name of the data type to look for in *conn_or_curs*
The connection or cursor passed to the function will be used to query the
database and look for the OID of the :sql:`json` type (or an alternative
type if *name* if provided). No query is performed if *oid* and *array_oid*
are provided. Raise `~psycopg2.ProgrammingError` if the type is not found.
"""
if oid is None:
oid, array_oid = _get_json_oids(conn_or_curs, name)
JSON, JSONARRAY = _create_json_typecasters(
oid, array_oid, loads=loads, name=name.upper())
register_type(JSON, not globally and conn_or_curs or None)
if JSONARRAY is not None:
register_type(JSONARRAY, not globally and conn_or_curs or None)
return JSON, JSONARRAY
def register_default_json(conn_or_curs=None, globally=False, loads=None):
"""
Create and register :sql:`json` typecasters for PostgreSQL 9.2 and following.
Since PostgreSQL 9.2 :sql:`json` is a builtin type, hence its oid is known
and fixed. This function allows specifying a customized *loads* function
for the default :sql:`json` type without querying the database.
All the parameters have the same meaning of `register_json()`.
"""
return register_json(conn_or_curs=conn_or_curs, globally=globally,
loads=loads, oid=JSON_OID, array_oid=JSONARRAY_OID)
def register_default_jsonb(conn_or_curs=None, globally=False, loads=None):
"""
Create and register :sql:`jsonb` typecasters for PostgreSQL 9.4 and following.
As in `register_default_json()`, the function allows to register a
customized *loads* function for the :sql:`jsonb` type at its known oid for
PostgreSQL 9.4 and following versions. All the parameters have the same
meaning of `register_json()`.
"""
return register_json(conn_or_curs=conn_or_curs, globally=globally,
loads=loads, oid=JSONB_OID, array_oid=JSONBARRAY_OID, name='jsonb')
def _create_json_typecasters(oid, array_oid, loads=None, name='JSON'):
"""Create typecasters for json data type."""
if loads is None:
loads = json.loads
def typecast_json(s, cur):
if s is None:
return None
return loads(s)
JSON = new_type((oid, ), name, typecast_json)
if array_oid is not None:
JSONARRAY = new_array_type((array_oid, ), "%sARRAY" % name, JSON)
else:
JSONARRAY = None
return JSON, JSONARRAY
def _get_json_oids(conn_or_curs, name='json'):
# lazy imports
from psycopg2.extensions import STATUS_IN_TRANSACTION
from psycopg2.extras import _solve_conn_curs
conn, curs = _solve_conn_curs(conn_or_curs)
# Store the transaction status of the connection to revert it after use
conn_status = conn.status
# column typarray not available before PG 8.3
typarray = conn.info.server_version >= 80300 and "typarray" or "NULL"
# get the oid for the hstore
curs.execute(
"SELECT t.oid, %s FROM pg_type t WHERE t.typname = %%s;"
% typarray, (name,))
r = curs.fetchone()
# revert the status of the connection as before the command
if conn_status != STATUS_IN_TRANSACTION and not conn.autocommit:
conn.rollback()
if not r:
raise conn.ProgrammingError("%s data type not found" % name)
return r
FILE:assets/driver/psycopg2/_lru_cache.py
"""
LRU cache implementation for Python 2.7
Ported from http://code.activestate.com/recipes/578078/ and simplified for our
use (only support maxsize > 0 and positional arguments).
"""
from collections import namedtuple
from functools import update_wrapper
from threading import RLock
_CacheInfo = namedtuple("CacheInfo", ["hits", "misses", "maxsize", "currsize"])
def lru_cache(maxsize=100):
"""Least-recently-used cache decorator.
Arguments to the cached function must be hashable.
See: http://en.wikipedia.org/wiki/Cache_algorithms#Least_Recently_Used
"""
def decorating_function(user_function):
cache = dict()
stats = [0, 0] # make statistics updateable non-locally
HITS, MISSES = 0, 1 # names for the stats fields
cache_get = cache.get # bound method to lookup key or return None
_len = len # localize the global len() function
lock = RLock() # linkedlist updates aren't threadsafe
root = [] # root of the circular doubly linked list
root[:] = [root, root, None, None] # initialize by pointing to self
nonlocal_root = [root] # make updateable non-locally
PREV, NEXT, KEY, RESULT = 0, 1, 2, 3 # names for the link fields
assert maxsize and maxsize > 0, "maxsize %s not supported" % maxsize
def wrapper(*args):
# size limited caching that tracks accesses by recency
key = args
with lock:
link = cache_get(key)
if link is not None:
# record recent use of the key by moving it to the
# front of the list
root, = nonlocal_root
link_prev, link_next, key, result = link
link_prev[NEXT] = link_next
link_next[PREV] = link_prev
last = root[PREV]
last[NEXT] = root[PREV] = link
link[PREV] = last
link[NEXT] = root
stats[HITS] += 1
return result
result = user_function(*args)
with lock:
root, = nonlocal_root
if key in cache:
# getting here means that this same key was added to the
# cache while the lock was released. since the link
# update is already done, we need only return the
# computed result and update the count of misses.
pass
elif _len(cache) >= maxsize:
# use the old root to store the new key and result
oldroot = root
oldroot[KEY] = key
oldroot[RESULT] = result
# empty the oldest link and make it the new root
root = nonlocal_root[0] = oldroot[NEXT]
oldkey = root[KEY]
# oldvalue = root[RESULT]
root[KEY] = root[RESULT] = None
# now update the cache dictionary for the new links
del cache[oldkey]
cache[key] = oldroot
else:
# put result in a new link at the front of the list
last = root[PREV]
link = [last, root, key, result]
last[NEXT] = root[PREV] = cache[key] = link
stats[MISSES] += 1
return result
def cache_info():
"""Report cache statistics"""
with lock:
return _CacheInfo(stats[HITS], stats[MISSES], maxsize, len(cache))
def cache_clear():
"""Clear the cache and cache statistics"""
with lock:
cache.clear()
root = nonlocal_root[0]
root[:] = [root, root, None, None]
stats[:] = [0, 0]
wrapper.__wrapped__ = user_function
wrapper.cache_info = cache_info
wrapper.cache_clear = cache_clear
return update_wrapper(wrapper, user_function)
return decorating_function
FILE:assets/driver/psycopg2/_range.py
"""Implementation of the Range type and adaptation
"""
# psycopg/_range.py - Implementation of the Range type and adaptation
#
# Copyright (C) 2012-2019 Daniele Varrazzo <[email protected]>
# Copyright (C) 2020 The Psycopg Team
#
# psycopg2 is free software: you can redistribute it and/or modify it
# under the terms of the GNU Lesser General Public License as published
# by the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# In addition, as a special exception, the copyright holders give
# permission to link this program with the OpenSSL library (or with
# modified versions of OpenSSL that use the same license as OpenSSL),
# and distribute linked combinations including the two.
#
# You must obey the GNU Lesser General Public License in all respects for
# all of the code used other than OpenSSL.
#
# psycopg2 is distributed in the hope that it will be useful, but WITHOUT
# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
# FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public
# License for more details.
import re
from psycopg2._psycopg import ProgrammingError, InterfaceError
from psycopg2.extensions import ISQLQuote, adapt, register_adapter
from psycopg2.extensions import new_type, new_array_type, register_type
from psycopg2.compat import string_types
class Range(object):
"""Python representation for a PostgreSQL |range|_ type.
:param lower: lower bound for the range. `!None` means unbound
:param upper: upper bound for the range. `!None` means unbound
:param bounds: one of the literal strings ``()``, ``[)``, ``(]``, ``[]``,
representing whether the lower or upper bounds are included
:param empty: if `!True`, the range is empty
"""
__slots__ = ('_lower', '_upper', '_bounds')
def __init__(self, lower=None, upper=None, bounds='[)', empty=False):
if not empty:
if bounds not in ('[)', '(]', '()', '[]'):
raise ValueError("bound flags not valid: %r" % bounds)
self._lower = lower
self._upper = upper
self._bounds = bounds
else:
self._lower = self._upper = self._bounds = None
def __repr__(self):
if self._bounds is None:
return "%s(empty=True)" % self.__class__.__name__
else:
return "%s(%r, %r, %r)" % (self.__class__.__name__,
self._lower, self._upper, self._bounds)
def __str__(self):
if self._bounds is None:
return 'empty'
items = [
self._bounds[0],
str(self._lower),
', ',
str(self._upper),
self._bounds[1]
]
return ''.join(items)
@property
def lower(self):
"""The lower bound of the range. `!None` if empty or unbound."""
return self._lower
@property
def upper(self):
"""The upper bound of the range. `!None` if empty or unbound."""
return self._upper
@property
def isempty(self):
"""`!True` if the range is empty."""
return self._bounds is None
@property
def lower_inf(self):
"""`!True` if the range doesn't have a lower bound."""
if self._bounds is None:
return False
return self._lower is None
@property
def upper_inf(self):
"""`!True` if the range doesn't have an upper bound."""
if self._bounds is None:
return False
return self._upper is None
@property
def lower_inc(self):
"""`!True` if the lower bound is included in the range."""
if self._bounds is None or self._lower is None:
return False
return self._bounds[0] == '['
@property
def upper_inc(self):
"""`!True` if the upper bound is included in the range."""
if self._bounds is None or self._upper is None:
return False
return self._bounds[1] == ']'
def __contains__(self, x):
if self._bounds is None:
return False
if self._lower is not None:
if self._bounds[0] == '[':
if x < self._lower:
return False
else:
if x <= self._lower:
return False
if self._upper is not None:
if self._bounds[1] == ']':
if x > self._upper:
return False
else:
if x >= self._upper:
return False
return True
def __bool__(self):
return self._bounds is not None
def __nonzero__(self):
# Python 2 compatibility
return type(self).__bool__(self)
def __eq__(self, other):
if not isinstance(other, Range):
return False
return (self._lower == other._lower
and self._upper == other._upper
and self._bounds == other._bounds)
def __ne__(self, other):
return not self.__eq__(other)
def __hash__(self):
return hash((self._lower, self._upper, self._bounds))
# as the postgres docs describe for the server-side stuff,
# ordering is rather arbitrary, but will remain stable
# and consistent.
def __lt__(self, other):
if not isinstance(other, Range):
return NotImplemented
for attr in ('_lower', '_upper', '_bounds'):
self_value = getattr(self, attr)
other_value = getattr(other, attr)
if self_value == other_value:
pass
elif self_value is None:
return True
elif other_value is None:
return False
else:
return self_value < other_value
return False
def __le__(self, other):
if self == other:
return True
else:
return self.__lt__(other)
def __gt__(self, other):
if isinstance(other, Range):
return other.__lt__(self)
else:
return NotImplemented
def __ge__(self, other):
if self == other:
return True
else:
return self.__gt__(other)
def __getstate__(self):
return {slot: getattr(self, slot)
for slot in self.__slots__ if hasattr(self, slot)}
def __setstate__(self, state):
for slot, value in state.items():
setattr(self, slot, value)
def register_range(pgrange, pyrange, conn_or_curs, globally=False):
"""Create and register an adapter and the typecasters to convert between
a PostgreSQL |range|_ type and a PostgreSQL `Range` subclass.
:param pgrange: the name of the PostgreSQL |range| type. Can be
schema-qualified
:param pyrange: a `Range` strict subclass, or just a name to give to a new
class
:param conn_or_curs: a connection or cursor used to find the oid of the
range and its subtype; the typecaster is registered in a scope limited
to this object, unless *globally* is set to `!True`
:param globally: if `!False` (default) register the typecaster only on
*conn_or_curs*, otherwise register it globally
:return: `RangeCaster` instance responsible for the conversion
If a string is passed to *pyrange*, a new `Range` subclass is created
with such name and will be available as the `~RangeCaster.range` attribute
of the returned `RangeCaster` object.
The function queries the database on *conn_or_curs* to inspect the
*pgrange* type and raises `~psycopg2.ProgrammingError` if the type is not
found. If querying the database is not advisable, use directly the
`RangeCaster` class and register the adapter and typecasters using the
provided functions.
"""
caster = RangeCaster._from_db(pgrange, pyrange, conn_or_curs)
caster._register(not globally and conn_or_curs or None)
return caster
class RangeAdapter(object):
"""`ISQLQuote` adapter for `Range` subclasses.
This is an abstract class: concrete classes must set a `name` class
attribute or override `getquoted()`.
"""
name = None
def __init__(self, adapted):
self.adapted = adapted
def __conform__(self, proto):
if self._proto is ISQLQuote:
return self
def prepare(self, conn):
self._conn = conn
def getquoted(self):
if self.name is None:
raise NotImplementedError(
'RangeAdapter must be subclassed overriding its name '
'or the getquoted() method')
r = self.adapted
if r.isempty:
return b"'empty'::" + self.name.encode('utf8')
if r.lower is not None:
a = adapt(r.lower)
if hasattr(a, 'prepare'):
a.prepare(self._conn)
lower = a.getquoted()
else:
lower = b'NULL'
if r.upper is not None:
a = adapt(r.upper)
if hasattr(a, 'prepare'):
a.prepare(self._conn)
upper = a.getquoted()
else:
upper = b'NULL'
return self.name.encode('utf8') + b'(' + lower + b', ' + upper \
+ b", '" + r._bounds.encode('utf8') + b"')"
class RangeCaster(object):
"""Helper class to convert between `Range` and PostgreSQL range types.
Objects of this class are usually created by `register_range()`. Manual
creation could be useful if querying the database is not advisable: in
this case the oids must be provided.
"""
def __init__(self, pgrange, pyrange, oid, subtype_oid, array_oid=None):
self.subtype_oid = subtype_oid
self._create_ranges(pgrange, pyrange)
name = self.adapter.name or self.adapter.__class__.__name__
self.typecaster = new_type((oid,), name, self.parse)
if array_oid is not None:
self.array_typecaster = new_array_type(
(array_oid,), name + "ARRAY", self.typecaster)
else:
self.array_typecaster = None
def _create_ranges(self, pgrange, pyrange):
"""Create Range and RangeAdapter classes if needed."""
# if got a string create a new RangeAdapter concrete type (with a name)
# else take it as an adapter. Passing an adapter should be considered
# an implementation detail and is not documented. It is currently used
# for the numeric ranges.
self.adapter = None
if isinstance(pgrange, string_types):
self.adapter = type(pgrange, (RangeAdapter,), {})
self.adapter.name = pgrange
else:
try:
if issubclass(pgrange, RangeAdapter) \
and pgrange is not RangeAdapter:
self.adapter = pgrange
except TypeError:
pass
if self.adapter is None:
raise TypeError(
'pgrange must be a string or a RangeAdapter strict subclass')
self.range = None
try:
if isinstance(pyrange, string_types):
self.range = type(pyrange, (Range,), {})
if issubclass(pyrange, Range) and pyrange is not Range:
self.range = pyrange
except TypeError:
pass
if self.range is None:
raise TypeError(
'pyrange must be a type or a Range strict subclass')
@classmethod
def _from_db(self, name, pyrange, conn_or_curs):
"""Return a `RangeCaster` instance for the type *pgrange*.
Raise `ProgrammingError` if the type is not found.
"""
from psycopg2.extensions import STATUS_IN_TRANSACTION
from psycopg2.extras import _solve_conn_curs
conn, curs = _solve_conn_curs(conn_or_curs)
if conn.info.server_version < 90200:
raise ProgrammingError("range types not available in version %s"
% conn.info.server_version)
# Store the transaction status of the connection to revert it after use
conn_status = conn.status
# Use the correct schema
if '.' in name:
schema, tname = name.split('.', 1)
else:
tname = name
schema = 'public'
# get the type oid and attributes
try:
curs.execute("""\
select rngtypid, rngsubtype,
(select typarray from pg_type where oid = rngtypid)
from pg_range r
join pg_type t on t.oid = rngtypid
join pg_namespace ns on ns.oid = typnamespace
where typname = %s and ns.nspname = %s;
""", (tname, schema))
except ProgrammingError:
if not conn.autocommit:
conn.rollback()
raise
else:
rec = curs.fetchone()
# revert the status of the connection as before the command
if (conn_status != STATUS_IN_TRANSACTION
and not conn.autocommit):
conn.rollback()
if not rec:
raise ProgrammingError(
"PostgreSQL type '%s' not found" % name)
type, subtype, array = rec
return RangeCaster(name, pyrange,
oid=type, subtype_oid=subtype, array_oid=array)
_re_range = re.compile(r"""
( \(|\[ ) # lower bound flag
(?: # lower bound:
" ( (?: [^"] | "")* ) " # - a quoted string
| ( [^",]+ ) # - or an unquoted string
)? # - or empty (not catched)
,
(?: # upper bound:
" ( (?: [^"] | "")* ) " # - a quoted string
| ( [^"\)\]]+ ) # - or an unquoted string
)? # - or empty (not catched)
( \)|\] ) # upper bound flag
""", re.VERBOSE)
_re_undouble = re.compile(r'(["\\])\1')
def parse(self, s, cur=None):
if s is None:
return None
if s == 'empty':
return self.range(empty=True)
m = self._re_range.match(s)
if m is None:
raise InterfaceError("failed to parse range: '%s'" % s)
lower = m.group(3)
if lower is None:
lower = m.group(2)
if lower is not None:
lower = self._re_undouble.sub(r"\1", lower)
upper = m.group(5)
if upper is None:
upper = m.group(4)
if upper is not None:
upper = self._re_undouble.sub(r"\1", upper)
if cur is not None:
lower = cur.cast(self.subtype_oid, lower)
upper = cur.cast(self.subtype_oid, upper)
bounds = m.group(1) + m.group(6)
return self.range(lower, upper, bounds)
def _register(self, scope=None):
register_type(self.typecaster, scope)
if self.array_typecaster is not None:
register_type(self.array_typecaster, scope)
register_adapter(self.range, self.adapter)
class NumericRange(Range):
"""A `Range` suitable to pass Python numeric types to a PostgreSQL range.
PostgreSQL types :sql:`int4range`, :sql:`int8range`, :sql:`numrange` are
casted into `!NumericRange` instances.
"""
pass
class DateRange(Range):
"""Represents :sql:`daterange` values."""
pass
class DateTimeRange(Range):
"""Represents :sql:`tsrange` values."""
pass
class DateTimeTZRange(Range):
"""Represents :sql:`tstzrange` values."""
pass
# Special adaptation for NumericRange. Allows to pass number range regardless
# of whether they are ints, floats and what size of ints are, which are
# pointless in Python world. On the way back, no numeric range is casted to
# NumericRange, but only to their subclasses
class NumberRangeAdapter(RangeAdapter):
"""Adapt a range if the subtype doesn't need quotes."""
def getquoted(self):
r = self.adapted
if r.isempty:
return b"'empty'"
if not r.lower_inf:
# not exactly: we are relying that none of these object is really
# quoted (they are numbers). Also, I'm lazy and not preparing the
# adapter because I assume encoding doesn't matter for these
# objects.
lower = adapt(r.lower).getquoted().decode('ascii')
else:
lower = ''
if not r.upper_inf:
upper = adapt(r.upper).getquoted().decode('ascii')
else:
upper = ''
return ("'%s%s,%s%s'" % (
r._bounds[0], lower, upper, r._bounds[1])).encode('ascii')
# TO_DO: probably won't work with infs, nans and other tricky cases.
register_adapter(NumericRange, NumberRangeAdapter)
# Register globally typecasters and adapters for builtin range types.
# note: the adapter is registered more than once, but this is harmless.
int4range_caster = RangeCaster(NumberRangeAdapter, NumericRange,
oid=3904, subtype_oid=23, array_oid=3905)
int4range_caster._register()
int8range_caster = RangeCaster(NumberRangeAdapter, NumericRange,
oid=3926, subtype_oid=20, array_oid=3927)
int8range_caster._register()
numrange_caster = RangeCaster(NumberRangeAdapter, NumericRange,
oid=3906, subtype_oid=1700, array_oid=3907)
numrange_caster._register()
daterange_caster = RangeCaster('daterange', DateRange,
oid=3912, subtype_oid=1082, array_oid=3913)
daterange_caster._register()
tsrange_caster = RangeCaster('tsrange', DateTimeRange,
oid=3908, subtype_oid=1114, array_oid=3909)
tsrange_caster._register()
tstzrange_caster = RangeCaster('tstzrange', DateTimeTZRange,
oid=3910, subtype_oid=1184, array_oid=3911)
tstzrange_caster._register()
FILE:assets/driver/psycopg2/compat.py
import sys
__all__ = ['string_types', 'text_type', 'lru_cache']
if sys.version_info[0] == 2:
# Python 2
PY2 = True
PY3 = False
string_types = basestring,
text_type = unicode
from ._lru_cache import lru_cache
else:
# Python 3
PY2 = False
PY3 = True
string_types = str,
text_type = str
from functools import lru_cache
FILE:assets/driver/psycopg2/errorcodes.py
"""Error codes for PostgresSQL
This module contains symbolic names for all PostgreSQL error codes.
"""
# psycopg2/errorcodes.py - PostgreSQL error codes
#
# Copyright (C) 2006-2019 Johan Dahlin <[email protected]>
# Copyright (C) 2020 The Psycopg Team
#
# psycopg2 is free software: you can redistribute it and/or modify it
# under the terms of the GNU Lesser General Public License as published
# by the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# In addition, as a special exception, the copyright holders give
# permission to link this program with the OpenSSL library (or with
# modified versions of OpenSSL that use the same license as OpenSSL),
# and distribute linked combinations including the two.
#
# You must obey the GNU Lesser General Public License in all respects for
# all of the code used other than OpenSSL.
#
# psycopg2 is distributed in the hope that it will be useful, but WITHOUT
# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
# FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public
# License for more details.
#
# Based on:
#
# https://www.postgresql.org/docs/current/static/errcodes-appendix.html
#
def lookup(code, _cache={}):
"""Lookup an error code or class code and return its symbolic name.
Raise `KeyError` if the code is not found.
"""
if _cache:
return _cache[code]
# Generate the lookup map at first usage.
tmp = {}
for k, v in globals().items():
if isinstance(v, str) and len(v) in (2, 5):
# Strip trailing underscore used to disambiguate duplicate values
tmp[v] = k.rstrip("_")
assert tmp
# Atomic update, to avoid race condition on import (bug #382)
_cache.update(tmp)
return _cache[code]
# autogenerated data: do not edit below this point.
# Error classes
CLASS_SUCCESSFUL_COMPLETION = '00'
CLASS_WARNING = '01'
CLASS_NO_DATA = '02'
CLASS_SQL_STATEMENT_NOT_YET_COMPLETE = '03'
CLASS_CONNECTION_EXCEPTION = '08'
CLASS_TRIGGERED_ACTION_EXCEPTION = '09'
CLASS_FEATURE_NOT_SUPPORTED = '0A'
CLASS_INVALID_TRANSACTION_INITIATION = '0B'
CLASS_LOCATOR_EXCEPTION = '0F'
CLASS_INVALID_GRANTOR = '0L'
CLASS_INVALID_ROLE_SPECIFICATION = '0P'
CLASS_DIAGNOSTICS_EXCEPTION = '0Z'
CLASS_CASE_NOT_FOUND = '20'
CLASS_CARDINALITY_VIOLATION = '21'
CLASS_DATA_EXCEPTION = '22'
CLASS_INTEGRITY_CONSTRAINT_VIOLATION = '23'
CLASS_INVALID_CURSOR_STATE = '24'
CLASS_INVALID_TRANSACTION_STATE = '25'
CLASS_INVALID_SQL_STATEMENT_NAME = '26'
CLASS_TRIGGERED_DATA_CHANGE_VIOLATION = '27'
CLASS_INVALID_AUTHORIZATION_SPECIFICATION = '28'
CLASS_DEPENDENT_PRIVILEGE_DESCRIPTORS_STILL_EXIST = '2B'
CLASS_INVALID_TRANSACTION_TERMINATION = '2D'
CLASS_SQL_ROUTINE_EXCEPTION = '2F'
CLASS_INVALID_CURSOR_NAME = '34'
CLASS_EXTERNAL_ROUTINE_EXCEPTION = '38'
CLASS_EXTERNAL_ROUTINE_INVOCATION_EXCEPTION = '39'
CLASS_SAVEPOINT_EXCEPTION = '3B'
CLASS_INVALID_CATALOG_NAME = '3D'
CLASS_INVALID_SCHEMA_NAME = '3F'
CLASS_TRANSACTION_ROLLBACK = '40'
CLASS_SYNTAX_ERROR_OR_ACCESS_RULE_VIOLATION = '42'
CLASS_WITH_CHECK_OPTION_VIOLATION = '44'
CLASS_INSUFFICIENT_RESOURCES = '53'
CLASS_PROGRAM_LIMIT_EXCEEDED = '54'
CLASS_OBJECT_NOT_IN_PREREQUISITE_STATE = '55'
CLASS_OPERATOR_INTERVENTION = '57'
CLASS_SYSTEM_ERROR = '58'
CLASS_SNAPSHOT_FAILURE = '72'
CLASS_CONFIGURATION_FILE_ERROR = 'F0'
CLASS_FOREIGN_DATA_WRAPPER_ERROR = 'HV'
CLASS_PL_PGSQL_ERROR = 'P0'
CLASS_INTERNAL_ERROR = 'XX'
# Class 00 - Successful Completion
SUCCESSFUL_COMPLETION = '00000'
# Class 01 - Warning
WARNING = '01000'
NULL_VALUE_ELIMINATED_IN_SET_FUNCTION = '01003'
STRING_DATA_RIGHT_TRUNCATION_ = '01004'
PRIVILEGE_NOT_REVOKED = '01006'
PRIVILEGE_NOT_GRANTED = '01007'
IMPLICIT_ZERO_BIT_PADDING = '01008'
DYNAMIC_RESULT_SETS_RETURNED = '0100C'
DEPRECATED_FEATURE = '01P01'
# Class 02 - No Data (this is also a warning class per the SQL standard)
NO_DATA = '02000'
NO_ADDITIONAL_DYNAMIC_RESULT_SETS_RETURNED = '02001'
# Class 03 - SQL Statement Not Yet Complete
SQL_STATEMENT_NOT_YET_COMPLETE = '03000'
# Class 08 - Connection Exception
CONNECTION_EXCEPTION = '08000'
SQLCLIENT_UNABLE_TO_ESTABLISH_SQLCONNECTION = '08001'
CONNECTION_DOES_NOT_EXIST = '08003'
SQLSERVER_REJECTED_ESTABLISHMENT_OF_SQLCONNECTION = '08004'
CONNECTION_FAILURE = '08006'
TRANSACTION_RESOLUTION_UNKNOWN = '08007'
PROTOCOL_VIOLATION = '08P01'
# Class 09 - Triggered Action Exception
TRIGGERED_ACTION_EXCEPTION = '09000'
# Class 0A - Feature Not Supported
FEATURE_NOT_SUPPORTED = '0A000'
# Class 0B - Invalid Transaction Initiation
INVALID_TRANSACTION_INITIATION = '0B000'
# Class 0F - Locator Exception
LOCATOR_EXCEPTION = '0F000'
INVALID_LOCATOR_SPECIFICATION = '0F001'
# Class 0L - Invalid Grantor
INVALID_GRANTOR = '0L000'
INVALID_GRANT_OPERATION = '0LP01'
# Class 0P - Invalid Role Specification
INVALID_ROLE_SPECIFICATION = '0P000'
# Class 0Z - Diagnostics Exception
DIAGNOSTICS_EXCEPTION = '0Z000'
STACKED_DIAGNOSTICS_ACCESSED_WITHOUT_ACTIVE_HANDLER = '0Z002'
# Class 20 - Case Not Found
CASE_NOT_FOUND = '20000'
# Class 21 - Cardinality Violation
CARDINALITY_VIOLATION = '21000'
# Class 22 - Data Exception
DATA_EXCEPTION = '22000'
STRING_DATA_RIGHT_TRUNCATION = '22001'
NULL_VALUE_NO_INDICATOR_PARAMETER = '22002'
NUMERIC_VALUE_OUT_OF_RANGE = '22003'
NULL_VALUE_NOT_ALLOWED_ = '22004'
ERROR_IN_ASSIGNMENT = '22005'
INVALID_DATETIME_FORMAT = '22007'
DATETIME_FIELD_OVERFLOW = '22008'
INVALID_TIME_ZONE_DISPLACEMENT_VALUE = '22009'
ESCAPE_CHARACTER_CONFLICT = '2200B'
INVALID_USE_OF_ESCAPE_CHARACTER = '2200C'
INVALID_ESCAPE_OCTET = '2200D'
ZERO_LENGTH_CHARACTER_STRING = '2200F'
MOST_SPECIFIC_TYPE_MISMATCH = '2200G'
SEQUENCE_GENERATOR_LIMIT_EXCEEDED = '2200H'
NOT_AN_XML_DOCUMENT = '2200L'
INVALID_XML_DOCUMENT = '2200M'
INVALID_XML_CONTENT = '2200N'
INVALID_XML_COMMENT = '2200S'
INVALID_XML_PROCESSING_INSTRUCTION = '2200T'
INVALID_INDICATOR_PARAMETER_VALUE = '22010'
SUBSTRING_ERROR = '22011'
DIVISION_BY_ZERO = '22012'
INVALID_PRECEDING_OR_FOLLOWING_SIZE = '22013'
INVALID_ARGUMENT_FOR_NTILE_FUNCTION = '22014'
INTERVAL_FIELD_OVERFLOW = '22015'
INVALID_ARGUMENT_FOR_NTH_VALUE_FUNCTION = '22016'
INVALID_CHARACTER_VALUE_FOR_CAST = '22018'
INVALID_ESCAPE_CHARACTER = '22019'
INVALID_REGULAR_EXPRESSION = '2201B'
INVALID_ARGUMENT_FOR_LOGARITHM = '2201E'
INVALID_ARGUMENT_FOR_POWER_FUNCTION = '2201F'
INVALID_ARGUMENT_FOR_WIDTH_BUCKET_FUNCTION = '2201G'
INVALID_ROW_COUNT_IN_LIMIT_CLAUSE = '2201W'
INVALID_ROW_COUNT_IN_RESULT_OFFSET_CLAUSE = '2201X'
INVALID_LIMIT_VALUE = '22020'
CHARACTER_NOT_IN_REPERTOIRE = '22021'
INDICATOR_OVERFLOW = '22022'
INVALID_PARAMETER_VALUE = '22023'
UNTERMINATED_C_STRING = '22024'
INVALID_ESCAPE_SEQUENCE = '22025'
STRING_DATA_LENGTH_MISMATCH = '22026'
TRIM_ERROR = '22027'
ARRAY_SUBSCRIPT_ERROR = '2202E'
INVALID_TABLESAMPLE_REPEAT = '2202G'
INVALID_TABLESAMPLE_ARGUMENT = '2202H'
DUPLICATE_JSON_OBJECT_KEY_VALUE = '22030'
INVALID_ARGUMENT_FOR_SQL_JSON_DATETIME_FUNCTION = '22031'
INVALID_JSON_TEXT = '22032'
INVALID_SQL_JSON_SUBSCRIPT = '22033'
MORE_THAN_ONE_SQL_JSON_ITEM = '22034'
NO_SQL_JSON_ITEM = '22035'
NON_NUMERIC_SQL_JSON_ITEM = '22036'
NON_UNIQUE_KEYS_IN_A_JSON_OBJECT = '22037'
SINGLETON_SQL_JSON_ITEM_REQUIRED = '22038'
SQL_JSON_ARRAY_NOT_FOUND = '22039'
SQL_JSON_MEMBER_NOT_FOUND = '2203A'
SQL_JSON_NUMBER_NOT_FOUND = '2203B'
SQL_JSON_OBJECT_NOT_FOUND = '2203C'
TOO_MANY_JSON_ARRAY_ELEMENTS = '2203D'
TOO_MANY_JSON_OBJECT_MEMBERS = '2203E'
SQL_JSON_SCALAR_REQUIRED = '2203F'
FLOATING_POINT_EXCEPTION = '22P01'
INVALID_TEXT_REPRESENTATION = '22P02'
INVALID_BINARY_REPRESENTATION = '22P03'
BAD_COPY_FILE_FORMAT = '22P04'
UNTRANSLATABLE_CHARACTER = '22P05'
NONSTANDARD_USE_OF_ESCAPE_CHARACTER = '22P06'
# Class 23 - Integrity Constraint Violation
INTEGRITY_CONSTRAINT_VIOLATION = '23000'
RESTRICT_VIOLATION = '23001'
NOT_NULL_VIOLATION = '23502'
FOREIGN_KEY_VIOLATION = '23503'
UNIQUE_VIOLATION = '23505'
CHECK_VIOLATION = '23514'
EXCLUSION_VIOLATION = '23P01'
# Class 24 - Invalid Cursor State
INVALID_CURSOR_STATE = '24000'
# Class 25 - Invalid Transaction State
INVALID_TRANSACTION_STATE = '25000'
ACTIVE_SQL_TRANSACTION = '25001'
BRANCH_TRANSACTION_ALREADY_ACTIVE = '25002'
INAPPROPRIATE_ACCESS_MODE_FOR_BRANCH_TRANSACTION = '25003'
INAPPROPRIATE_ISOLATION_LEVEL_FOR_BRANCH_TRANSACTION = '25004'
NO_ACTIVE_SQL_TRANSACTION_FOR_BRANCH_TRANSACTION = '25005'
READ_ONLY_SQL_TRANSACTION = '25006'
SCHEMA_AND_DATA_STATEMENT_MIXING_NOT_SUPPORTED = '25007'
HELD_CURSOR_REQUIRES_SAME_ISOLATION_LEVEL = '25008'
NO_ACTIVE_SQL_TRANSACTION = '25P01'
IN_FAILED_SQL_TRANSACTION = '25P02'
IDLE_IN_TRANSACTION_SESSION_TIMEOUT = '25P03'
# Class 26 - Invalid SQL Statement Name
INVALID_SQL_STATEMENT_NAME = '26000'
# Class 27 - Triggered Data Change Violation
TRIGGERED_DATA_CHANGE_VIOLATION = '27000'
# Class 28 - Invalid Authorization Specification
INVALID_AUTHORIZATION_SPECIFICATION = '28000'
INVALID_PASSWORD = '28P01'
# Class 2B - Dependent Privilege Descriptors Still Exist
DEPENDENT_PRIVILEGE_DESCRIPTORS_STILL_EXIST = '2B000'
DEPENDENT_OBJECTS_STILL_EXIST = '2BP01'
# Class 2D - Invalid Transaction Termination
INVALID_TRANSACTION_TERMINATION = '2D000'
# Class 2F - SQL Routine Exception
SQL_ROUTINE_EXCEPTION = '2F000'
MODIFYING_SQL_DATA_NOT_PERMITTED_ = '2F002'
PROHIBITED_SQL_STATEMENT_ATTEMPTED_ = '2F003'
READING_SQL_DATA_NOT_PERMITTED_ = '2F004'
FUNCTION_EXECUTED_NO_RETURN_STATEMENT = '2F005'
# Class 34 - Invalid Cursor Name
INVALID_CURSOR_NAME = '34000'
# Class 38 - External Routine Exception
EXTERNAL_ROUTINE_EXCEPTION = '38000'
CONTAINING_SQL_NOT_PERMITTED = '38001'
MODIFYING_SQL_DATA_NOT_PERMITTED = '38002'
PROHIBITED_SQL_STATEMENT_ATTEMPTED = '38003'
READING_SQL_DATA_NOT_PERMITTED = '38004'
# Class 39 - External Routine Invocation Exception
EXTERNAL_ROUTINE_INVOCATION_EXCEPTION = '39000'
INVALID_SQLSTATE_RETURNED = '39001'
NULL_VALUE_NOT_ALLOWED = '39004'
TRIGGER_PROTOCOL_VIOLATED = '39P01'
SRF_PROTOCOL_VIOLATED = '39P02'
EVENT_TRIGGER_PROTOCOL_VIOLATED = '39P03'
# Class 3B - Savepoint Exception
SAVEPOINT_EXCEPTION = '3B000'
INVALID_SAVEPOINT_SPECIFICATION = '3B001'
# Class 3D - Invalid Catalog Name
INVALID_CATALOG_NAME = '3D000'
# Class 3F - Invalid Schema Name
INVALID_SCHEMA_NAME = '3F000'
# Class 40 - Transaction Rollback
TRANSACTION_ROLLBACK = '40000'
SERIALIZATION_FAILURE = '40001'
TRANSACTION_INTEGRITY_CONSTRAINT_VIOLATION = '40002'
STATEMENT_COMPLETION_UNKNOWN = '40003'
DEADLOCK_DETECTED = '40P01'
# Class 42 - Syntax Error or Access Rule Violation
SYNTAX_ERROR_OR_ACCESS_RULE_VIOLATION = '42000'
INSUFFICIENT_PRIVILEGE = '42501'
SYNTAX_ERROR = '42601'
INVALID_NAME = '42602'
INVALID_COLUMN_DEFINITION = '42611'
NAME_TOO_LONG = '42622'
DUPLICATE_COLUMN = '42701'
AMBIGUOUS_COLUMN = '42702'
UNDEFINED_COLUMN = '42703'
UNDEFINED_OBJECT = '42704'
DUPLICATE_OBJECT = '42710'
DUPLICATE_ALIAS = '42712'
DUPLICATE_FUNCTION = '42723'
AMBIGUOUS_FUNCTION = '42725'
GROUPING_ERROR = '42803'
DATATYPE_MISMATCH = '42804'
WRONG_OBJECT_TYPE = '42809'
INVALID_FOREIGN_KEY = '42830'
CANNOT_COERCE = '42846'
UNDEFINED_FUNCTION = '42883'
GENERATED_ALWAYS = '428C9'
RESERVED_NAME = '42939'
UNDEFINED_TABLE = '42P01'
UNDEFINED_PARAMETER = '42P02'
DUPLICATE_CURSOR = '42P03'
DUPLICATE_DATABASE = '42P04'
DUPLICATE_PREPARED_STATEMENT = '42P05'
DUPLICATE_SCHEMA = '42P06'
DUPLICATE_TABLE = '42P07'
AMBIGUOUS_PARAMETER = '42P08'
AMBIGUOUS_ALIAS = '42P09'
INVALID_COLUMN_REFERENCE = '42P10'
INVALID_CURSOR_DEFINITION = '42P11'
INVALID_DATABASE_DEFINITION = '42P12'
INVALID_FUNCTION_DEFINITION = '42P13'
INVALID_PREPARED_STATEMENT_DEFINITION = '42P14'
INVALID_SCHEMA_DEFINITION = '42P15'
INVALID_TABLE_DEFINITION = '42P16'
INVALID_OBJECT_DEFINITION = '42P17'
INDETERMINATE_DATATYPE = '42P18'
INVALID_RECURSION = '42P19'
WINDOWING_ERROR = '42P20'
COLLATION_MISMATCH = '42P21'
INDETERMINATE_COLLATION = '42P22'
# Class 44 - WITH CHECK OPTION Violation
WITH_CHECK_OPTION_VIOLATION = '44000'
# Class 53 - Insufficient Resources
INSUFFICIENT_RESOURCES = '53000'
DISK_FULL = '53100'
OUT_OF_MEMORY = '53200'
TOO_MANY_CONNECTIONS = '53300'
CONFIGURATION_LIMIT_EXCEEDED = '53400'
# Class 54 - Program Limit Exceeded
PROGRAM_LIMIT_EXCEEDED = '54000'
STATEMENT_TOO_COMPLEX = '54001'
TOO_MANY_COLUMNS = '54011'
TOO_MANY_ARGUMENTS = '54023'
# Class 55 - Object Not In Prerequisite State
OBJECT_NOT_IN_PREREQUISITE_STATE = '55000'
OBJECT_IN_USE = '55006'
CANT_CHANGE_RUNTIME_PARAM = '55P02'
LOCK_NOT_AVAILABLE = '55P03'
UNSAFE_NEW_ENUM_VALUE_USAGE = '55P04'
# Class 57 - Operator Intervention
OPERATOR_INTERVENTION = '57000'
QUERY_CANCELED = '57014'
ADMIN_SHUTDOWN = '57P01'
CRASH_SHUTDOWN = '57P02'
CANNOT_CONNECT_NOW = '57P03'
DATABASE_DROPPED = '57P04'
# Class 58 - System Error (errors external to PostgreSQL itself)
SYSTEM_ERROR = '58000'
IO_ERROR = '58030'
UNDEFINED_FILE = '58P01'
DUPLICATE_FILE = '58P02'
# Class 72 - Snapshot Failure
SNAPSHOT_TOO_OLD = '72000'
# Class F0 - Configuration File Error
CONFIG_FILE_ERROR = 'F0000'
LOCK_FILE_EXISTS = 'F0001'
# Class HV - Foreign Data Wrapper Error (SQL/MED)
FDW_ERROR = 'HV000'
FDW_OUT_OF_MEMORY = 'HV001'
FDW_DYNAMIC_PARAMETER_VALUE_NEEDED = 'HV002'
FDW_INVALID_DATA_TYPE = 'HV004'
FDW_COLUMN_NAME_NOT_FOUND = 'HV005'
FDW_INVALID_DATA_TYPE_DESCRIPTORS = 'HV006'
FDW_INVALID_COLUMN_NAME = 'HV007'
FDW_INVALID_COLUMN_NUMBER = 'HV008'
FDW_INVALID_USE_OF_NULL_POINTER = 'HV009'
FDW_INVALID_STRING_FORMAT = 'HV00A'
FDW_INVALID_HANDLE = 'HV00B'
FDW_INVALID_OPTION_INDEX = 'HV00C'
FDW_INVALID_OPTION_NAME = 'HV00D'
FDW_OPTION_NAME_NOT_FOUND = 'HV00J'
FDW_REPLY_HANDLE = 'HV00K'
FDW_UNABLE_TO_CREATE_EXECUTION = 'HV00L'
FDW_UNABLE_TO_CREATE_REPLY = 'HV00M'
FDW_UNABLE_TO_ESTABLISH_CONNECTION = 'HV00N'
FDW_NO_SCHEMAS = 'HV00P'
FDW_SCHEMA_NOT_FOUND = 'HV00Q'
FDW_TABLE_NOT_FOUND = 'HV00R'
FDW_FUNCTION_SEQUENCE_ERROR = 'HV010'
FDW_TOO_MANY_HANDLES = 'HV014'
FDW_INCONSISTENT_DESCRIPTOR_INFORMATION = 'HV021'
FDW_INVALID_ATTRIBUTE_VALUE = 'HV024'
FDW_INVALID_STRING_LENGTH_OR_BUFFER_LENGTH = 'HV090'
FDW_INVALID_DESCRIPTOR_FIELD_IDENTIFIER = 'HV091'
# Class P0 - PL/pgSQL Error
PLPGSQL_ERROR = 'P0000'
RAISE_EXCEPTION = 'P0001'
NO_DATA_FOUND = 'P0002'
TOO_MANY_ROWS = 'P0003'
ASSERT_FAILURE = 'P0004'
# Class XX - Internal Error
INTERNAL_ERROR = 'XX000'
DATA_CORRUPTED = 'XX001'
INDEX_CORRUPTED = 'XX002'
FILE:assets/driver/psycopg2/errors.py
"""Error classes for PostgreSQL error codes
"""
# psycopg/errors.py - SQLSTATE and DB-API exceptions
#
# Copyright (C) 2018-2019 Daniele Varrazzo <[email protected]>
# Copyright (C) 2020 The Psycopg Team
#
# psycopg2 is free software: you can redistribute it and/or modify it
# under the terms of the GNU Lesser General Public License as published
# by the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# In addition, as a special exception, the copyright holders give
# permission to link this program with the OpenSSL library (or with
# modified versions of OpenSSL that use the same license as OpenSSL),
# and distribute linked combinations including the two.
#
# You must obey the GNU Lesser General Public License in all respects for
# all of the code used other than OpenSSL.
#
# psycopg2 is distributed in the hope that it will be useful, but WITHOUT
# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
# FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public
# License for more details.
#
# NOTE: the exceptions are injected into this module by the C extention.
#
def lookup(code):
"""Lookup an error code and return its exception class.
Raise `!KeyError` if the code is not found.
"""
from psycopg2._psycopg import sqlstate_errors # avoid circular import
return sqlstate_errors[code]
FILE:assets/driver/psycopg2/extensions.py
"""psycopg extensions to the DBAPI-2.0
This module holds all the extensions to the DBAPI-2.0 provided by psycopg.
- `connection` -- the new-type inheritable connection class
- `cursor` -- the new-type inheritable cursor class
- `lobject` -- the new-type inheritable large object class
- `adapt()` -- exposes the PEP-246_ compatible adapting mechanism used
by psycopg to adapt Python types to PostgreSQL ones
.. _PEP-246: https://www.python.org/dev/peps/pep-0246/
"""
# psycopg/extensions.py - DBAPI-2.0 extensions specific to psycopg
#
# Copyright (C) 2003-2019 Federico Di Gregorio <[email protected]>
# Copyright (C) 2020 The Psycopg Team
#
# psycopg2 is free software: you can redistribute it and/or modify it
# under the terms of the GNU Lesser General Public License as published
# by the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# In addition, as a special exception, the copyright holders give
# permission to link this program with the OpenSSL library (or with
# modified versions of OpenSSL that use the same license as OpenSSL),
# and distribute linked combinations including the two.
#
# You must obey the GNU Lesser General Public License in all respects for
# all of the code used other than OpenSSL.
#
# psycopg2 is distributed in the hope that it will be useful, but WITHOUT
# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
# FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public
# License for more details.
import re as _re
from psycopg2._psycopg import ( # noqa
BINARYARRAY, BOOLEAN, BOOLEANARRAY, BYTES, BYTESARRAY, DATE, DATEARRAY,
DATETIMEARRAY, DECIMAL, DECIMALARRAY, FLOAT, FLOATARRAY, INTEGER,
INTEGERARRAY, INTERVAL, INTERVALARRAY, LONGINTEGER, LONGINTEGERARRAY,
ROWIDARRAY, STRINGARRAY, TIME, TIMEARRAY, UNICODE, UNICODEARRAY,
AsIs, Binary, Boolean, Float, Int, QuotedString, )
try:
from psycopg2._psycopg import ( # noqa
MXDATE, MXDATETIME, MXDATETIMETZ, MXINTERVAL, MXTIME, MXDATEARRAY,
MXDATETIMEARRAY, MXDATETIMETZARRAY, MXINTERVALARRAY, MXTIMEARRAY,
DateFromMx, TimeFromMx, TimestampFromMx, IntervalFromMx, )
except ImportError:
pass
from psycopg2._psycopg import ( # noqa
PYDATE, PYDATETIME, PYDATETIMETZ, PYINTERVAL, PYTIME, PYDATEARRAY,
PYDATETIMEARRAY, PYDATETIMETZARRAY, PYINTERVALARRAY, PYTIMEARRAY,
DateFromPy, TimeFromPy, TimestampFromPy, IntervalFromPy, )
from psycopg2._psycopg import ( # noqa
adapt, adapters, encodings, connection, cursor,
lobject, Xid, libpq_version, parse_dsn, quote_ident,
string_types, binary_types, new_type, new_array_type, register_type,
ISQLQuote, Notify, Diagnostics, Column, ConnectionInfo,
QueryCanceledError, TransactionRollbackError,
set_wait_callback, get_wait_callback, encrypt_password, )
"""Isolation level values."""
ISOLATION_LEVEL_AUTOCOMMIT = 0
ISOLATION_LEVEL_READ_UNCOMMITTED = 4
ISOLATION_LEVEL_READ_COMMITTED = 1
ISOLATION_LEVEL_REPEATABLE_READ = 2
ISOLATION_LEVEL_SERIALIZABLE = 3
ISOLATION_LEVEL_DEFAULT = None
"""psycopg connection status values."""
STATUS_SETUP = 0
STATUS_READY = 1
STATUS_BEGIN = 2
STATUS_SYNC = 3 # currently unused
STATUS_ASYNC = 4 # currently unused
STATUS_PREPARED = 5
# This is a useful mnemonic to check if the connection is in a transaction
STATUS_IN_TRANSACTION = STATUS_BEGIN
"""psycopg asynchronous connection polling values"""
POLL_OK = 0
POLL_READ = 1
POLL_WRITE = 2
POLL_ERROR = 3
"""Backend transaction status values."""
TRANSACTION_STATUS_IDLE = 0
TRANSACTION_STATUS_ACTIVE = 1
TRANSACTION_STATUS_INTRANS = 2
TRANSACTION_STATUS_INERROR = 3
TRANSACTION_STATUS_UNKNOWN = 4
def register_adapter(typ, callable):
"""Register 'callable' as an ISQLQuote adapter for type 'typ'."""
adapters[(typ, ISQLQuote)] = callable
# The SQL_IN class is the official adapter for tuples starting from 2.0.6.
class SQL_IN(object):
"""Adapt any iterable to an SQL quotable object."""
def __init__(self, seq):
self._seq = seq
self._conn = None
def prepare(self, conn):
self._conn = conn
def getquoted(self):
# this is the important line: note how every object in the
# list is adapted and then how getquoted() is called on it
pobjs = [adapt(o) for o in self._seq]
if self._conn is not None:
for obj in pobjs:
if hasattr(obj, 'prepare'):
obj.prepare(self._conn)
qobjs = [o.getquoted() for o in pobjs]
return b'(' + b', '.join(qobjs) + b')'
def __str__(self):
return str(self.getquoted())
class NoneAdapter(object):
"""Adapt None to NULL.
This adapter is not used normally as a fast path in mogrify uses NULL,
but it makes easier to adapt composite types.
"""
def __init__(self, obj):
pass
def getquoted(self, _null=b"NULL"):
return _null
def make_dsn(dsn=None, **kwargs):
"""Convert a set of keywords into a connection strings."""
if dsn is None and not kwargs:
return ''
# If no kwarg is specified don't mung the dsn, but verify it
if not kwargs:
parse_dsn(dsn)
return dsn
# Override the dsn with the parameters
if 'database' in kwargs:
if 'dbname' in kwargs:
raise TypeError(
"you can't specify both 'database' and 'dbname' arguments")
kwargs['dbname'] = kwargs.pop('database')
# Drop the None arguments
kwargs = {k: v for (k, v) in kwargs.items() if v is not None}
if dsn is not None:
tmp = parse_dsn(dsn)
tmp.update(kwargs)
kwargs = tmp
dsn = " ".join(["%s=%s" % (k, _param_escape(str(v)))
for (k, v) in kwargs.items()])
# verify that the returned dsn is valid
parse_dsn(dsn)
return dsn
def _param_escape(s,
re_escape=_re.compile(r"([\\'])"),
re_space=_re.compile(r'\s')):
"""
Apply the escaping rule required by PQconnectdb
"""
if not s:
return "''"
s = re_escape.sub(r'\\\1', s)
if re_space.search(s):
s = "'" + s + "'"
return s
# Create default json typecasters for PostgreSQL 9.2 oids
from psycopg2._json import register_default_json, register_default_jsonb # noqa
try:
JSON, JSONARRAY = register_default_json()
JSONB, JSONBARRAY = register_default_jsonb()
except ImportError:
pass
del register_default_json, register_default_jsonb
# Create default Range typecasters
from psycopg2. _range import Range # noqa
del Range
# Add the "cleaned" version of the encodings to the key.
# When the encoding is set its name is cleaned up from - and _ and turned
# uppercase, so an encoding not respecting these rules wouldn't be found in the
# encodings keys and would raise an exception with the unicode typecaster
for k, v in list(encodings.items()):
k = k.replace('_', '').replace('-', '').upper()
encodings[k] = v
del k, v
FILE:assets/driver/psycopg2/extras.py
"""Miscellaneous goodies for psycopg2
This module is a generic place used to hold little helper functions
and classes until a better place in the distribution is found.
"""
# psycopg/extras.py - miscellaneous extra goodies for psycopg
#
# Copyright (C) 2003-2019 Federico Di Gregorio <[email protected]>
# Copyright (C) 2020 The Psycopg Team
#
# psycopg2 is free software: you can redistribute it and/or modify it
# under the terms of the GNU Lesser General Public License as published
# by the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# In addition, as a special exception, the copyright holders give
# permission to link this program with the OpenSSL library (or with
# modified versions of OpenSSL that use the same license as OpenSSL),
# and distribute linked combinations including the two.
#
# You must obey the GNU Lesser General Public License in all respects for
# all of the code used other than OpenSSL.
#
# psycopg2 is distributed in the hope that it will be useful, but WITHOUT
# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
# FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public
# License for more details.
import os as _os
import time as _time
import re as _re
from collections import namedtuple, OrderedDict
import logging as _logging
import psycopg2
from psycopg2 import extensions as _ext
from .extensions import cursor as _cursor
from .extensions import connection as _connection
from .extensions import adapt as _A, quote_ident
from .compat import PY2, PY3, lru_cache
from psycopg2._psycopg import ( # noqa
REPLICATION_PHYSICAL, REPLICATION_LOGICAL,
ReplicationConnection as _replicationConnection,
ReplicationCursor as _replicationCursor,
ReplicationMessage)
# expose the json adaptation stuff into the module
from psycopg2._json import ( # noqa
json, Json, register_json, register_default_json, register_default_jsonb)
# Expose range-related objects
from psycopg2._range import ( # noqa
Range, NumericRange, DateRange, DateTimeRange, DateTimeTZRange,
register_range, RangeAdapter, RangeCaster)
# Expose ipaddress-related objects
from psycopg2._ipaddress import register_ipaddress # noqa
class DictCursorBase(_cursor):
"""Base class for all dict-like cursors."""
def __init__(self, *args, **kwargs):
if 'row_factory' in kwargs:
row_factory = kwargs['row_factory']
del kwargs['row_factory']
else:
raise NotImplementedError(
"DictCursorBase can't be instantiated without a row factory.")
super(DictCursorBase, self).__init__(*args, **kwargs)
self._query_executed = False
self._prefetch = False
self.row_factory = row_factory
def fetchone(self):
if self._prefetch:
res = super(DictCursorBase, self).fetchone()
if self._query_executed:
self._build_index()
if not self._prefetch:
res = super(DictCursorBase, self).fetchone()
return res
def fetchmany(self, size=None):
if self._prefetch:
res = super(DictCursorBase, self).fetchmany(size)
if self._query_executed:
self._build_index()
if not self._prefetch:
res = super(DictCursorBase, self).fetchmany(size)
return res
def fetchall(self):
if self._prefetch:
res = super(DictCursorBase, self).fetchall()
if self._query_executed:
self._build_index()
if not self._prefetch:
res = super(DictCursorBase, self).fetchall()
return res
def __iter__(self):
try:
if self._prefetch:
res = super(DictCursorBase, self).__iter__()
first = next(res)
if self._query_executed:
self._build_index()
if not self._prefetch:
res = super(DictCursorBase, self).__iter__()
first = next(res)
yield first
while True:
yield next(res)
except StopIteration:
return
class DictConnection(_connection):
"""A connection that uses `DictCursor` automatically."""
def cursor(self, *args, **kwargs):
kwargs.setdefault('cursor_factory', self.cursor_factory or DictCursor)
return super(DictConnection, self).cursor(*args, **kwargs)
class DictCursor(DictCursorBase):
"""A cursor that keeps a list of column name -> index mappings__.
.. __: https://docs.python.org/glossary.html#term-mapping
"""
def __init__(self, *args, **kwargs):
kwargs['row_factory'] = DictRow
super(DictCursor, self).__init__(*args, **kwargs)
self._prefetch = True
def execute(self, query, vars=None):
self.index = OrderedDict()
self._query_executed = True
return super(DictCursor, self).execute(query, vars)
def callproc(self, procname, vars=None):
self.index = OrderedDict()
self._query_executed = True
return super(DictCursor, self).callproc(procname, vars)
def _build_index(self):
if self._query_executed and self.description:
for i in range(len(self.description)):
self.index[self.description[i][0]] = i
self._query_executed = False
class DictRow(list):
"""A row object that allow by-column-name access to data."""
__slots__ = ('_index',)
def __init__(self, cursor):
self._index = cursor.index
self[:] = [None] * len(cursor.description)
def __getitem__(self, x):
if not isinstance(x, (int, slice)):
x = self._index[x]
return super(DictRow, self).__getitem__(x)
def __setitem__(self, x, v):
if not isinstance(x, (int, slice)):
x = self._index[x]
super(DictRow, self).__setitem__(x, v)
def items(self):
g = super(DictRow, self).__getitem__
return ((n, g(self._index[n])) for n in self._index)
def keys(self):
return iter(self._index)
def values(self):
g = super(DictRow, self).__getitem__
return (g(self._index[n]) for n in self._index)
def get(self, x, default=None):
try:
return self[x]
except Exception:
return default
def copy(self):
return OrderedDict(self.items())
def __contains__(self, x):
return x in self._index
def __reduce__(self):
# this is apparently useless, but it fixes #1073
return super(DictRow, self).__reduce__()
def __getstate__(self):
return self[:], self._index.copy()
def __setstate__(self, data):
self[:] = data[0]
self._index = data[1]
if PY2:
iterkeys = keys
itervalues = values
iteritems = items
has_key = __contains__
def keys(self):
return list(self.iterkeys())
def values(self):
return tuple(self.itervalues())
def items(self):
return list(self.iteritems())
class RealDictConnection(_connection):
"""A connection that uses `RealDictCursor` automatically."""
def cursor(self, *args, **kwargs):
kwargs.setdefault('cursor_factory', self.cursor_factory or RealDictCursor)
return super(RealDictConnection, self).cursor(*args, **kwargs)
class RealDictCursor(DictCursorBase):
"""A cursor that uses a real dict as the base type for rows.
Note that this cursor is extremely specialized and does not allow
the normal access (using integer indices) to fetched data. If you need
to access database rows both as a dictionary and a list, then use
the generic `DictCursor` instead of `!RealDictCursor`.
"""
def __init__(self, *args, **kwargs):
kwargs['row_factory'] = RealDictRow
super(RealDictCursor, self).__init__(*args, **kwargs)
def execute(self, query, vars=None):
self.column_mapping = []
self._query_executed = True
return super(RealDictCursor, self).execute(query, vars)
def callproc(self, procname, vars=None):
self.column_mapping = []
self._query_executed = True
return super(RealDictCursor, self).callproc(procname, vars)
def _build_index(self):
if self._query_executed and self.description:
self.column_mapping = [d[0] for d in self.description]
self._query_executed = False
class RealDictRow(OrderedDict):
"""A `!dict` subclass representing a data record."""
def __init__(self, *args, **kwargs):
if args and isinstance(args[0], _cursor):
cursor = args[0]
args = args[1:]
else:
cursor = None
super(RealDictRow, self).__init__(*args, **kwargs)
if cursor is not None:
# Required for named cursors
if cursor.description and not cursor.column_mapping:
cursor._build_index()
# Store the cols mapping in the dict itself until the row is fully
# populated, so we don't need to add attributes to the class
# (hence keeping its maintenance, special pickle support, etc.)
self[RealDictRow] = cursor.column_mapping
def __setitem__(self, key, value):
if RealDictRow in self:
# We are in the row building phase
mapping = self[RealDictRow]
super(RealDictRow, self).__setitem__(mapping[key], value)
if key == len(mapping) - 1:
# Row building finished
del self[RealDictRow]
return
super(RealDictRow, self).__setitem__(key, value)
class NamedTupleConnection(_connection):
"""A connection that uses `NamedTupleCursor` automatically."""
def cursor(self, *args, **kwargs):
kwargs.setdefault('cursor_factory', self.cursor_factory or NamedTupleCursor)
return super(NamedTupleConnection, self).cursor(*args, **kwargs)
class NamedTupleCursor(_cursor):
"""A cursor that generates results as `~collections.namedtuple`.
`!fetch*()` methods will return named tuples instead of regular tuples, so
their elements can be accessed both as regular numeric items as well as
attributes.
>>> nt_cur = conn.cursor(cursor_factory=psycopg2.extras.NamedTupleCursor)
>>> rec = nt_cur.fetchone()
>>> rec
Record(id=1, num=100, data="abc'def")
>>> rec[1]
100
>>> rec.data
"abc'def"
"""
Record = None
MAX_CACHE = 1024
def execute(self, query, vars=None):
self.Record = None
return super(NamedTupleCursor, self).execute(query, vars)
def executemany(self, query, vars):
self.Record = None
return super(NamedTupleCursor, self).executemany(query, vars)
def callproc(self, procname, vars=None):
self.Record = None
return super(NamedTupleCursor, self).callproc(procname, vars)
def fetchone(self):
t = super(NamedTupleCursor, self).fetchone()
if t is not None:
nt = self.Record
if nt is None:
nt = self.Record = self._make_nt()
return nt._make(t)
def fetchmany(self, size=None):
ts = super(NamedTupleCursor, self).fetchmany(size)
nt = self.Record
if nt is None:
nt = self.Record = self._make_nt()
return list(map(nt._make, ts))
def fetchall(self):
ts = super(NamedTupleCursor, self).fetchall()
nt = self.Record
if nt is None:
nt = self.Record = self._make_nt()
return list(map(nt._make, ts))
def __iter__(self):
try:
it = super(NamedTupleCursor, self).__iter__()
t = next(it)
nt = self.Record
if nt is None:
nt = self.Record = self._make_nt()
yield nt._make(t)
while True:
yield nt._make(next(it))
except StopIteration:
return
# ascii except alnum and underscore
_re_clean = _re.compile(
'[' + _re.escape(' !"#$%&\'()*+,-./:;<=>?@[\\]^`{|}~') + ']')
def _make_nt(self):
key = tuple(d[0] for d in self.description) if self.description else ()
return self._cached_make_nt(key)
@classmethod
def _do_make_nt(cls, key):
fields = []
for s in key:
s = cls._re_clean.sub('_', s)
# Python identifier cannot start with numbers, namedtuple fields
# cannot start with underscore. So...
if s[0] == '_' or '0' <= s[0] <= '9':
s = 'f' + s
fields.append(s)
nt = namedtuple("Record", fields)
return nt
@lru_cache(512)
def _cached_make_nt(cls, key):
return cls._do_make_nt(key)
# Exposed for testability, and if someone wants to monkeypatch to tweak
# the cache size.
NamedTupleCursor._cached_make_nt = classmethod(_cached_make_nt)
class LoggingConnection(_connection):
"""A connection that logs all queries to a file or logger__ object.
.. __: https://docs.python.org/library/logging.html
"""
def initialize(self, logobj):
"""Initialize the connection to log to `!logobj`.
The `!logobj` parameter can be an open file object or a Logger/LoggerAdapter
instance from the standard logging module.
"""
self._logobj = logobj
if _logging and isinstance(
logobj, (_logging.Logger, _logging.LoggerAdapter)):
self.log = self._logtologger
else:
self.log = self._logtofile
def filter(self, msg, curs):
"""Filter the query before logging it.
This is the method to overwrite to filter unwanted queries out of the
log or to add some extra data to the output. The default implementation
just does nothing.
"""
return msg
def _logtofile(self, msg, curs):
msg = self.filter(msg, curs)
if msg:
if PY3 and isinstance(msg, bytes):
msg = msg.decode(_ext.encodings[self.encoding], 'replace')
self._logobj.write(msg + _os.linesep)
def _logtologger(self, msg, curs):
msg = self.filter(msg, curs)
if msg:
self._logobj.debug(msg)
def _check(self):
if not hasattr(self, '_logobj'):
raise self.ProgrammingError(
"LoggingConnection object has not been initialize()d")
def cursor(self, *args, **kwargs):
self._check()
kwargs.setdefault('cursor_factory', self.cursor_factory or LoggingCursor)
return super(LoggingConnection, self).cursor(*args, **kwargs)
class LoggingCursor(_cursor):
"""A cursor that logs queries using its connection logging facilities."""
def execute(self, query, vars=None):
try:
return super(LoggingCursor, self).execute(query, vars)
finally:
self.connection.log(self.query, self)
def callproc(self, procname, vars=None):
try:
return super(LoggingCursor, self).callproc(procname, vars)
finally:
self.connection.log(self.query, self)
class MinTimeLoggingConnection(LoggingConnection):
"""A connection that logs queries based on execution time.
This is just an example of how to sub-class `LoggingConnection` to
provide some extra filtering for the logged queries. Both the
`initialize()` and `filter()` methods are overwritten to make sure
that only queries executing for more than ``mintime`` ms are logged.
Note that this connection uses the specialized cursor
`MinTimeLoggingCursor`.
"""
def initialize(self, logobj, mintime=0):
LoggingConnection.initialize(self, logobj)
self._mintime = mintime
def filter(self, msg, curs):
t = (_time.time() - curs.timestamp) * 1000
if t > self._mintime:
if PY3 and isinstance(msg, bytes):
msg = msg.decode(_ext.encodings[self.encoding], 'replace')
return msg + _os.linesep + " (execution time: %d ms)" % t
def cursor(self, *args, **kwargs):
kwargs.setdefault('cursor_factory',
self.cursor_factory or MinTimeLoggingCursor)
return LoggingConnection.cursor(self, *args, **kwargs)
class MinTimeLoggingCursor(LoggingCursor):
"""The cursor sub-class companion to `MinTimeLoggingConnection`."""
def execute(self, query, vars=None):
self.timestamp = _time.time()
return LoggingCursor.execute(self, query, vars)
def callproc(self, procname, vars=None):
self.timestamp = _time.time()
return LoggingCursor.callproc(self, procname, vars)
class LogicalReplicationConnection(_replicationConnection):
def __init__(self, *args, **kwargs):
kwargs['replication_type'] = REPLICATION_LOGICAL
super(LogicalReplicationConnection, self).__init__(*args, **kwargs)
class PhysicalReplicationConnection(_replicationConnection):
def __init__(self, *args, **kwargs):
kwargs['replication_type'] = REPLICATION_PHYSICAL
super(PhysicalReplicationConnection, self).__init__(*args, **kwargs)
class StopReplication(Exception):
"""
Exception used to break out of the endless loop in
`~ReplicationCursor.consume_stream()`.
Subclass of `~exceptions.Exception`. Intentionally *not* inherited from
`~psycopg2.Error` as occurrence of this exception does not indicate an
error.
"""
pass
class ReplicationCursor(_replicationCursor):
"""A cursor used for communication on replication connections."""
def create_replication_slot(self, slot_name, slot_type=None, output_plugin=None):
"""Create streaming replication slot."""
command = "CREATE_REPLICATION_SLOT %s " % quote_ident(slot_name, self)
if slot_type is None:
slot_type = self.connection.replication_type
if slot_type == REPLICATION_LOGICAL:
if output_plugin is None:
raise psycopg2.ProgrammingError(
"output plugin name is required to create "
"logical replication slot")
command += "LOGICAL %s" % quote_ident(output_plugin, self)
elif slot_type == REPLICATION_PHYSICAL:
if output_plugin is not None:
raise psycopg2.ProgrammingError(
"cannot specify output plugin name when creating "
"physical replication slot")
command += "PHYSICAL"
else:
raise psycopg2.ProgrammingError(
"unrecognized replication type: %s" % repr(slot_type))
self.execute(command)
def drop_replication_slot(self, slot_name):
"""Drop streaming replication slot."""
command = "DROP_REPLICATION_SLOT %s" % quote_ident(slot_name, self)
self.execute(command)
def start_replication(
self, slot_name=None, slot_type=None, start_lsn=0,
timeline=0, options=None, decode=False, status_interval=10):
"""Start replication stream."""
command = "START_REPLICATION "
if slot_type is None:
slot_type = self.connection.replication_type
if slot_type == REPLICATION_LOGICAL:
if slot_name:
command += "SLOT %s " % quote_ident(slot_name, self)
else:
raise psycopg2.ProgrammingError(
"slot name is required for logical replication")
command += "LOGICAL "
elif slot_type == REPLICATION_PHYSICAL:
if slot_name:
command += "SLOT %s " % quote_ident(slot_name, self)
# don't add "PHYSICAL", before 9.4 it was just START_REPLICATION XXX/XXX
else:
raise psycopg2.ProgrammingError(
"unrecognized replication type: %s" % repr(slot_type))
if type(start_lsn) is str:
lsn = start_lsn.split('/')
lsn = "%X/%08X" % (int(lsn[0], 16), int(lsn[1], 16))
else:
lsn = "%X/%08X" % ((start_lsn >> 32) & 0xFFFFFFFF,
start_lsn & 0xFFFFFFFF)
command += lsn
if timeline != 0:
if slot_type == REPLICATION_LOGICAL:
raise psycopg2.ProgrammingError(
"cannot specify timeline for logical replication")
command += " TIMELINE %d" % timeline
if options:
if slot_type == REPLICATION_PHYSICAL:
raise psycopg2.ProgrammingError(
"cannot specify output plugin options for physical replication")
command += " ("
for k, v in options.items():
if not command.endswith('('):
command += ", "
command += "%s %s" % (quote_ident(k, self), _A(str(v)))
command += ")"
self.start_replication_expert(
command, decode=decode, status_interval=status_interval)
# allows replication cursors to be used in select.select() directly
def fileno(self):
return self.connection.fileno()
# a dbtype and adapter for Python UUID type
class UUID_adapter(object):
"""Adapt Python's uuid.UUID__ type to PostgreSQL's uuid__.
.. __: https://docs.python.org/library/uuid.html
.. __: https://www.postgresql.org/docs/current/static/datatype-uuid.html
"""
def __init__(self, uuid):
self._uuid = uuid
def __conform__(self, proto):
if proto is _ext.ISQLQuote:
return self
def getquoted(self):
return ("'%s'::uuid" % self._uuid).encode('utf8')
def __str__(self):
return "'%s'::uuid" % self._uuid
def register_uuid(oids=None, conn_or_curs=None):
"""Create the UUID type and an uuid.UUID adapter.
:param oids: oid for the PostgreSQL :sql:`uuid` type, or 2-items sequence
with oids of the type and the array. If not specified, use PostgreSQL
standard oids.
:param conn_or_curs: where to register the typecaster. If not specified,
register it globally.
"""
import uuid
if not oids:
oid1 = 2950
oid2 = 2951
elif isinstance(oids, (list, tuple)):
oid1, oid2 = oids
else:
oid1 = oids
oid2 = 2951
_ext.UUID = _ext.new_type((oid1, ), "UUID",
lambda data, cursor: data and uuid.UUID(data) or None)
_ext.UUIDARRAY = _ext.new_array_type((oid2,), "UUID[]", _ext.UUID)
_ext.register_type(_ext.UUID, conn_or_curs)
_ext.register_type(_ext.UUIDARRAY, conn_or_curs)
_ext.register_adapter(uuid.UUID, UUID_adapter)
return _ext.UUID
# a type, dbtype and adapter for PostgreSQL inet type
class Inet(object):
"""Wrap a string to allow for correct SQL-quoting of inet values.
Note that this adapter does NOT check the passed value to make
sure it really is an inet-compatible address but DOES call adapt()
on it to make sure it is impossible to execute an SQL-injection
by passing an evil value to the initializer.
"""
def __init__(self, addr):
self.addr = addr
def __repr__(self):
return "%s(%r)" % (self.__class__.__name__, self.addr)
def prepare(self, conn):
self._conn = conn
def getquoted(self):
obj = _A(self.addr)
if hasattr(obj, 'prepare'):
obj.prepare(self._conn)
return obj.getquoted() + b"::inet"
def __conform__(self, proto):
if proto is _ext.ISQLQuote:
return self
def __str__(self):
return str(self.addr)
def register_inet(oid=None, conn_or_curs=None):
"""Create the INET type and an Inet adapter.
:param oid: oid for the PostgreSQL :sql:`inet` type, or 2-items sequence
with oids of the type and the array. If not specified, use PostgreSQL
standard oids.
:param conn_or_curs: where to register the typecaster. If not specified,
register it globally.
"""
import warnings
warnings.warn(
"the inet adapter is deprecated, it's not very useful",
DeprecationWarning)
if not oid:
oid1 = 869
oid2 = 1041
elif isinstance(oid, (list, tuple)):
oid1, oid2 = oid
else:
oid1 = oid
oid2 = 1041
_ext.INET = _ext.new_type((oid1, ), "INET",
lambda data, cursor: data and Inet(data) or None)
_ext.INETARRAY = _ext.new_array_type((oid2, ), "INETARRAY", _ext.INET)
_ext.register_type(_ext.INET, conn_or_curs)
_ext.register_type(_ext.INETARRAY, conn_or_curs)
return _ext.INET
def wait_select(conn):
"""Wait until a connection or cursor has data available.
The function is an example of a wait callback to be registered with
`~psycopg2.extensions.set_wait_callback()`. This function uses
:py:func:`~select.select()` to wait for data to become available, and
therefore is able to handle/receive SIGINT/KeyboardInterrupt.
"""
import select
from psycopg2.extensions import POLL_OK, POLL_READ, POLL_WRITE
while True:
try:
state = conn.poll()
if state == POLL_OK:
break
elif state == POLL_READ:
select.select([conn.fileno()], [], [])
elif state == POLL_WRITE:
select.select([], [conn.fileno()], [])
else:
raise conn.OperationalError("bad state from poll: %s" % state)
except KeyboardInterrupt:
conn.cancel()
# the loop will be broken by a server error
continue
def _solve_conn_curs(conn_or_curs):
"""Return the connection and a DBAPI cursor from a connection or cursor."""
if conn_or_curs is None:
raise psycopg2.ProgrammingError("no connection or cursor provided")
if hasattr(conn_or_curs, 'execute'):
conn = conn_or_curs.connection
curs = conn.cursor(cursor_factory=_cursor)
else:
conn = conn_or_curs
curs = conn.cursor(cursor_factory=_cursor)
return conn, curs
class HstoreAdapter(object):
"""Adapt a Python dict to the hstore syntax."""
def __init__(self, wrapped):
self.wrapped = wrapped
def prepare(self, conn):
self.conn = conn
# use an old-style getquoted implementation if required
if conn.info.server_version < 90000:
self.getquoted = self._getquoted_8
def _getquoted_8(self):
"""Use the operators available in PG pre-9.0."""
if not self.wrapped:
return b"''::hstore"
adapt = _ext.adapt
rv = []
for k, v in self.wrapped.items():
k = adapt(k)
k.prepare(self.conn)
k = k.getquoted()
if v is not None:
v = adapt(v)
v.prepare(self.conn)
v = v.getquoted()
else:
v = b'NULL'
# XXX this b'ing is painfully inefficient!
rv.append(b"(" + k + b" => " + v + b")")
return b"(" + b'||'.join(rv) + b")"
def _getquoted_9(self):
"""Use the hstore(text[], text[]) function."""
if not self.wrapped:
return b"''::hstore"
k = _ext.adapt(list(self.wrapped.keys()))
k.prepare(self.conn)
v = _ext.adapt(list(self.wrapped.values()))
v.prepare(self.conn)
return b"hstore(" + k.getquoted() + b", " + v.getquoted() + b")"
getquoted = _getquoted_9
_re_hstore = _re.compile(r"""
# hstore key:
# a string of normal or escaped chars
"((?: [^"\\] | \\. )*)"
\s*=>\s* # hstore value
(?:
NULL # the value can be null - not catched
# or a quoted string like the key
| "((?: [^"\\] | \\. )*)"
)
(?:\s*,\s*|$) # pairs separated by comma or end of string.
""", _re.VERBOSE)
@classmethod
def parse(self, s, cur, _bsdec=_re.compile(r"\\(.)")):
"""Parse an hstore representation in a Python string.
The hstore is represented as something like::
"a"=>"1", "b"=>"2"
with backslash-escaped strings.
"""
if s is None:
return None
rv = {}
start = 0
for m in self._re_hstore.finditer(s):
if m is None or m.start() != start:
raise psycopg2.InterfaceError(
"error parsing hstore pair at char %d" % start)
k = _bsdec.sub(r'\1', m.group(1))
v = m.group(2)
if v is not None:
v = _bsdec.sub(r'\1', v)
rv[k] = v
start = m.end()
if start < len(s):
raise psycopg2.InterfaceError(
"error parsing hstore: unparsed data after char %d" % start)
return rv
@classmethod
def parse_unicode(self, s, cur):
"""Parse an hstore returning unicode keys and values."""
if s is None:
return None
s = s.decode(_ext.encodings[cur.connection.encoding])
return self.parse(s, cur)
@classmethod
def get_oids(self, conn_or_curs):
"""Return the lists of OID of the hstore and hstore[] types.
"""
conn, curs = _solve_conn_curs(conn_or_curs)
# Store the transaction status of the connection to revert it after use
conn_status = conn.status
# column typarray not available before PG 8.3
typarray = conn.info.server_version >= 80300 and "typarray" or "NULL"
rv0, rv1 = [], []
# get the oid for the hstore
curs.execute("""\
SELECT t.oid, %s
FROM pg_type t JOIN pg_namespace ns
ON typnamespace = ns.oid
WHERE typname = 'hstore';
""" % typarray)
for oids in curs:
rv0.append(oids[0])
rv1.append(oids[1])
# revert the status of the connection as before the command
if (conn_status != _ext.STATUS_IN_TRANSACTION
and not conn.autocommit):
conn.rollback()
return tuple(rv0), tuple(rv1)
def register_hstore(conn_or_curs, globally=False, unicode=False,
oid=None, array_oid=None):
r"""Register adapter and typecaster for `!dict`\-\ |hstore| conversions.
:param conn_or_curs: a connection or cursor: the typecaster will be
registered only on this object unless *globally* is set to `!True`
:param globally: register the adapter globally, not only on *conn_or_curs*
:param unicode: if `!True`, keys and values returned from the database
will be `!unicode` instead of `!str`. The option is not available on
Python 3
:param oid: the OID of the |hstore| type if known. If not, it will be
queried on *conn_or_curs*.
:param array_oid: the OID of the |hstore| array type if known. If not, it
will be queried on *conn_or_curs*.
The connection or cursor passed to the function will be used to query the
database and look for the OID of the |hstore| type (which may be different
across databases). If querying is not desirable (e.g. with
:ref:`asynchronous connections <async-support>`) you may specify it in the
*oid* parameter, which can be found using a query such as :sql:`SELECT
'hstore'::regtype::oid`. Analogously you can obtain a value for *array_oid*
using a query such as :sql:`SELECT 'hstore[]'::regtype::oid`.
Note that, when passing a dictionary from Python to the database, both
strings and unicode keys and values are supported. Dictionaries returned
from the database have keys/values according to the *unicode* parameter.
The |hstore| contrib module must be already installed in the database
(executing the ``hstore.sql`` script in your ``contrib`` directory).
Raise `~psycopg2.ProgrammingError` if the type is not found.
"""
if oid is None:
oid = HstoreAdapter.get_oids(conn_or_curs)
if oid is None or not oid[0]:
raise psycopg2.ProgrammingError(
"hstore type not found in the database. "
"please install it from your 'contrib/hstore.sql' file")
else:
array_oid = oid[1]
oid = oid[0]
if isinstance(oid, int):
oid = (oid,)
if array_oid is not None:
if isinstance(array_oid, int):
array_oid = (array_oid,)
else:
array_oid = tuple([x for x in array_oid if x])
# create and register the typecaster
if PY2 and unicode:
cast = HstoreAdapter.parse_unicode
else:
cast = HstoreAdapter.parse
HSTORE = _ext.new_type(oid, "HSTORE", cast)
_ext.register_type(HSTORE, not globally and conn_or_curs or None)
_ext.register_adapter(dict, HstoreAdapter)
if array_oid:
HSTOREARRAY = _ext.new_array_type(array_oid, "HSTOREARRAY", HSTORE)
_ext.register_type(HSTOREARRAY, not globally and conn_or_curs or None)
class CompositeCaster(object):
"""Helps conversion of a PostgreSQL composite type into a Python object.
The class is usually created by the `register_composite()` function.
You may want to create and register manually instances of the class if
querying the database at registration time is not desirable (such as when
using an :ref:`asynchronous connections <async-support>`).
"""
def __init__(self, name, oid, attrs, array_oid=None, schema=None):
self.name = name
self.schema = schema
self.oid = oid
self.array_oid = array_oid
self.attnames = [a[0] for a in attrs]
self.atttypes = [a[1] for a in attrs]
self._create_type(name, self.attnames)
self.typecaster = _ext.new_type((oid,), name, self.parse)
if array_oid:
self.array_typecaster = _ext.new_array_type(
(array_oid,), "%sARRAY" % name, self.typecaster)
else:
self.array_typecaster = None
def parse(self, s, curs):
if s is None:
return None
tokens = self.tokenize(s)
if len(tokens) != len(self.atttypes):
raise psycopg2.DataError(
"expecting %d components for the type %s, %d found instead" %
(len(self.atttypes), self.name, len(tokens)))
values = [curs.cast(oid, token)
for oid, token in zip(self.atttypes, tokens)]
return self.make(values)
def make(self, values):
"""Return a new Python object representing the data being casted.
*values* is the list of attributes, already casted into their Python
representation.
You can subclass this method to :ref:`customize the composite cast
<custom-composite>`.
"""
return self._ctor(values)
_re_tokenize = _re.compile(r"""
\(? ([,)]) # an empty token, representing NULL
| \(? " ((?: [^"] | "")*) " [,)] # or a quoted string
| \(? ([^",)]+) [,)] # or an unquoted string
""", _re.VERBOSE)
_re_undouble = _re.compile(r'(["\\])\1')
@classmethod
def tokenize(self, s):
rv = []
for m in self._re_tokenize.finditer(s):
if m is None:
raise psycopg2.InterfaceError("can't parse type: %r" % s)
if m.group(1) is not None:
rv.append(None)
elif m.group(2) is not None:
rv.append(self._re_undouble.sub(r"\1", m.group(2)))
else:
rv.append(m.group(3))
return rv
def _create_type(self, name, attnames):
self.type = namedtuple(name, attnames)
self._ctor = self.type._make
@classmethod
def _from_db(self, name, conn_or_curs):
"""Return a `CompositeCaster` instance for the type *name*.
Raise `ProgrammingError` if the type is not found.
"""
conn, curs = _solve_conn_curs(conn_or_curs)
# Store the transaction status of the connection to revert it after use
conn_status = conn.status
# Use the correct schema
if '.' in name:
schema, tname = name.split('.', 1)
else:
tname = name
schema = 'public'
# column typarray not available before PG 8.3
typarray = conn.info.server_version >= 80300 and "typarray" or "NULL"
# get the type oid and attributes
curs.execute("""\
SELECT t.oid, %s, attname, atttypid
FROM pg_type t
JOIN pg_namespace ns ON typnamespace = ns.oid
JOIN pg_attribute a ON attrelid = typrelid
WHERE typname = %%s AND nspname = %%s
AND attnum > 0 AND NOT attisdropped
ORDER BY attnum;
""" % typarray, (tname, schema))
recs = curs.fetchall()
# revert the status of the connection as before the command
if (conn_status != _ext.STATUS_IN_TRANSACTION
and not conn.autocommit):
conn.rollback()
if not recs:
raise psycopg2.ProgrammingError(
"PostgreSQL type '%s' not found" % name)
type_oid = recs[0][0]
array_oid = recs[0][1]
type_attrs = [(r[2], r[3]) for r in recs]
return self(tname, type_oid, type_attrs,
array_oid=array_oid, schema=schema)
def register_composite(name, conn_or_curs, globally=False, factory=None):
"""Register a typecaster to convert a composite type into a tuple.
:param name: the name of a PostgreSQL composite type, e.g. created using
the |CREATE TYPE|_ command
:param conn_or_curs: a connection or cursor used to find the type oid and
components; the typecaster is registered in a scope limited to this
object, unless *globally* is set to `!True`
:param globally: if `!False` (default) register the typecaster only on
*conn_or_curs*, otherwise register it globally
:param factory: if specified it should be a `CompositeCaster` subclass: use
it to :ref:`customize how to cast composite types <custom-composite>`
:return: the registered `CompositeCaster` or *factory* instance
responsible for the conversion
"""
if factory is None:
factory = CompositeCaster
caster = factory._from_db(name, conn_or_curs)
_ext.register_type(caster.typecaster, not globally and conn_or_curs or None)
if caster.array_typecaster is not None:
_ext.register_type(
caster.array_typecaster, not globally and conn_or_curs or None)
return caster
def _paginate(seq, page_size):
"""Consume an iterable and return it in chunks.
Every chunk is at most `page_size`. Never return an empty chunk.
"""
page = []
it = iter(seq)
while True:
try:
for i in range(page_size):
page.append(next(it))
yield page
page = []
except StopIteration:
if page:
yield page
return
def execute_batch(cur, sql, argslist, page_size=100):
r"""Execute groups of statements in fewer server roundtrips.
Execute *sql* several times, against all parameters set (sequences or
mappings) found in *argslist*.
The function is semantically similar to
.. parsed-literal::
*cur*\.\ `~cursor.executemany`\ (\ *sql*\ , *argslist*\ )
but has a different implementation: Psycopg will join the statements into
fewer multi-statement commands, each one containing at most *page_size*
statements, resulting in a reduced number of server roundtrips.
After the execution of the function the `cursor.rowcount` property will
**not** contain a total result.
"""
for page in _paginate(argslist, page_size=page_size):
sqls = [cur.mogrify(sql, args) for args in page]
cur.execute(b";".join(sqls))
def execute_values(cur, sql, argslist, template=None, page_size=100, fetch=False):
'''Execute a statement using :sql:`VALUES` with a sequence of parameters.
:param cur: the cursor to use to execute the query.
:param sql: the query to execute. It must contain a single ``%s``
placeholder, which will be replaced by a `VALUES list`__.
Example: ``"INSERT INTO mytable (id, f1, f2) VALUES %s"``.
:param argslist: sequence of sequences or dictionaries with the arguments
to send to the query. The type and content must be consistent with
*template*.
:param template: the snippet to merge to every item in *argslist* to
compose the query.
- If the *argslist* items are sequences it should contain positional
placeholders (e.g. ``"(%s, %s, %s)"``, or ``"(%s, %s, 42)``" if there
are constants value...).
- If the *argslist* items are mappings it should contain named
placeholders (e.g. ``"(%(id)s, %(f1)s, 42)"``).
If not specified, assume the arguments are sequence and use a simple
positional template (i.e. ``(%s, %s, ...)``), with the number of
placeholders sniffed by the first element in *argslist*.
:param page_size: maximum number of *argslist* items to include in every
statement. If there are more items the function will execute more than
one statement.
:param fetch: if `!True` return the query results into a list (like in a
`~cursor.fetchall()`). Useful for queries with :sql:`RETURNING`
clause.
.. __: https://www.postgresql.org/docs/current/static/queries-values.html
After the execution of the function the `cursor.rowcount` property will
**not** contain a total result.
While :sql:`INSERT` is an obvious candidate for this function it is
possible to use it with other statements, for example::
>>> cur.execute(
... "create table test (id int primary key, v1 int, v2 int)")
>>> execute_values(cur,
... "INSERT INTO test (id, v1, v2) VALUES %s",
... [(1, 2, 3), (4, 5, 6), (7, 8, 9)])
>>> execute_values(cur,
... """UPDATE test SET v1 = data.v1 FROM (VALUES %s) AS data (id, v1)
... WHERE test.id = data.id""",
... [(1, 20), (4, 50)])
>>> cur.execute("select * from test order by id")
>>> cur.fetchall()
[(1, 20, 3), (4, 50, 6), (7, 8, 9)])
'''
from psycopg2.sql import Composable
if isinstance(sql, Composable):
sql = sql.as_string(cur)
# we can't just use sql % vals because vals is bytes: if sql is bytes
# there will be some decoding error because of stupid codec used, and Py3
# doesn't implement % on bytes.
if not isinstance(sql, bytes):
sql = sql.encode(_ext.encodings[cur.connection.encoding])
pre, post = _split_sql(sql)
result = [] if fetch else None
for page in _paginate(argslist, page_size=page_size):
if template is None:
template = b'(' + b','.join([b'%s'] * len(page[0])) + b')'
parts = pre[:]
for args in page:
parts.append(cur.mogrify(template, args))
parts.append(b',')
parts[-1:] = post
cur.execute(b''.join(parts))
if fetch:
result.extend(cur.fetchall())
return result
def _split_sql(sql):
"""Split *sql* on a single ``%s`` placeholder.
Split on the %s, perform %% replacement and return pre, post lists of
snippets.
"""
curr = pre = []
post = []
tokens = _re.split(br'(%.)', sql)
for token in tokens:
if len(token) != 2 or token[:1] != b'%':
curr.append(token)
continue
if token[1:] == b's':
if curr is pre:
curr = post
else:
raise ValueError(
"the query contains more than one '%s' placeholder")
elif token[1:] == b'%':
curr.append(b'%')
else:
raise ValueError("unsupported format character: '%s'"
% token[1:].decode('ascii', 'replace'))
if curr is pre:
raise ValueError("the query doesn't contain any '%s' placeholder")
return pre, post
FILE:assets/driver/psycopg2/pool.py
"""Connection pooling for psycopg2
This module implements thread-safe (and not) connection pools.
"""
# psycopg/pool.py - pooling code for psycopg
#
# Copyright (C) 2003-2019 Federico Di Gregorio <[email protected]>
# Copyright (C) 2020 The Psycopg Team
#
# psycopg2 is free software: you can redistribute it and/or modify it
# under the terms of the GNU Lesser General Public License as published
# by the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# In addition, as a special exception, the copyright holders give
# permission to link this program with the OpenSSL library (or with
# modified versions of OpenSSL that use the same license as OpenSSL),
# and distribute linked combinations including the two.
#
# You must obey the GNU Lesser General Public License in all respects for
# all of the code used other than OpenSSL.
#
# psycopg2 is distributed in the hope that it will be useful, but WITHOUT
# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
# FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public
# License for more details.
import psycopg2
from psycopg2 import extensions as _ext
class PoolError(psycopg2.Error):
pass
class AbstractConnectionPool(object):
"""Generic key-based pooling code."""
def __init__(self, minconn, maxconn, *args, **kwargs):
"""Initialize the connection pool.
New 'minconn' connections are created immediately calling 'connfunc'
with given parameters. The connection pool will support a maximum of
about 'maxconn' connections.
"""
self.minconn = int(minconn)
self.maxconn = int(maxconn)
self.closed = False
self._args = args
self._kwargs = kwargs
self._pool = []
self._used = {}
self._rused = {} # id(conn) -> key map
self._keys = 0
for i in range(self.minconn):
self._connect()
def _connect(self, key=None):
"""Create a new connection and assign it to 'key' if not None."""
conn = psycopg2.connect(*self._args, **self._kwargs)
if key is not None:
self._used[key] = conn
self._rused[id(conn)] = key
else:
self._pool.append(conn)
return conn
def _getkey(self):
"""Return a new unique key."""
self._keys += 1
return self._keys
def _getconn(self, key=None):
"""Get a free connection and assign it to 'key' if not None."""
if self.closed:
raise PoolError("connection pool is closed")
if key is None:
key = self._getkey()
if key in self._used:
return self._used[key]
if self._pool:
self._used[key] = conn = self._pool.pop()
self._rused[id(conn)] = key
return conn
else:
if len(self._used) == self.maxconn:
raise PoolError("connection pool exhausted")
return self._connect(key)
def _putconn(self, conn, key=None, close=False):
"""Put away a connection."""
if self.closed:
raise PoolError("connection pool is closed")
if key is None:
key = self._rused.get(id(conn))
if key is None:
raise PoolError("trying to put unkeyed connection")
if len(self._pool) < self.minconn and not close:
# Return the connection into a consistent state before putting
# it back into the pool
if not conn.closed:
status = conn.info.transaction_status
if status == _ext.TRANSACTION_STATUS_UNKNOWN:
# server connection lost
conn.close()
elif status != _ext.TRANSACTION_STATUS_IDLE:
# connection in error or in transaction
conn.rollback()
self._pool.append(conn)
else:
# regular idle connection
self._pool.append(conn)
# If the connection is closed, we just discard it.
else:
conn.close()
# here we check for the presence of key because it can happen that a
# thread tries to put back a connection after a call to close
if not self.closed or key in self._used:
del self._used[key]
del self._rused[id(conn)]
def _closeall(self):
"""Close all connections.
Note that this can lead to some code fail badly when trying to use
an already closed connection. If you call .closeall() make sure
your code can deal with it.
"""
if self.closed:
raise PoolError("connection pool is closed")
for conn in self._pool + list(self._used.values()):
try:
conn.close()
except Exception:
pass
self.closed = True
class SimpleConnectionPool(AbstractConnectionPool):
"""A connection pool that can't be shared across different threads."""
getconn = AbstractConnectionPool._getconn
putconn = AbstractConnectionPool._putconn
closeall = AbstractConnectionPool._closeall
class ThreadedConnectionPool(AbstractConnectionPool):
"""A connection pool that works with the threading module."""
def __init__(self, minconn, maxconn, *args, **kwargs):
"""Initialize the threading lock."""
import threading
AbstractConnectionPool.__init__(
self, minconn, maxconn, *args, **kwargs)
self._lock = threading.Lock()
def getconn(self, key=None):
"""Get a free connection and assign it to 'key' if not None."""
self._lock.acquire()
try:
return self._getconn(key)
finally:
self._lock.release()
def putconn(self, conn=None, key=None, close=False):
"""Put away an unused connection."""
self._lock.acquire()
try:
self._putconn(conn, key, close)
finally:
self._lock.release()
def closeall(self):
"""Close all connections (even the one currently in use.)"""
self._lock.acquire()
try:
self._closeall()
finally:
self._lock.release()
FILE:assets/driver/psycopg2/sql.py
"""SQL composition utility module
"""
# psycopg/sql.py - SQL composition utility module
#
# Copyright (C) 2016-2019 Daniele Varrazzo <[email protected]>
# Copyright (C) 2020 The Psycopg Team
#
# psycopg2 is free software: you can redistribute it and/or modify it
# under the terms of the GNU Lesser General Public License as published
# by the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# In addition, as a special exception, the copyright holders give
# permission to link this program with the OpenSSL library (or with
# modified versions of OpenSSL that use the same license as OpenSSL),
# and distribute linked combinations including the two.
#
# You must obey the GNU Lesser General Public License in all respects for
# all of the code used other than OpenSSL.
#
# psycopg2 is distributed in the hope that it will be useful, but WITHOUT
# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
# FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public
# License for more details.
import string
from psycopg2 import extensions as ext
from psycopg2.compat import PY3, string_types
_formatter = string.Formatter()
class Composable(object):
"""
Abstract base class for objects that can be used to compose an SQL string.
`!Composable` objects can be passed directly to `~cursor.execute()`,
`~cursor.executemany()`, `~cursor.copy_expert()` in place of the query
string.
`!Composable` objects can be joined using the ``+`` operator: the result
will be a `Composed` instance containing the objects joined. The operator
``*`` is also supported with an integer argument: the result is a
`!Composed` instance containing the left argument repeated as many times as
requested.
"""
def __init__(self, wrapped):
self._wrapped = wrapped
def __repr__(self):
return "%s(%r)" % (self.__class__.__name__, self._wrapped)
def as_string(self, context):
"""
Return the string value of the object.
:param context: the context to evaluate the string into.
:type context: `connection` or `cursor`
The method is automatically invoked by `~cursor.execute()`,
`~cursor.executemany()`, `~cursor.copy_expert()` if a `!Composable` is
passed instead of the query string.
"""
raise NotImplementedError
def __add__(self, other):
if isinstance(other, Composed):
return Composed([self]) + other
if isinstance(other, Composable):
return Composed([self]) + Composed([other])
else:
return NotImplemented
def __mul__(self, n):
return Composed([self] * n)
def __eq__(self, other):
return type(self) is type(other) and self._wrapped == other._wrapped
def __ne__(self, other):
return not self.__eq__(other)
class Composed(Composable):
"""
A `Composable` object made of a sequence of `!Composable`.
The object is usually created using `!Composable` operators and methods.
However it is possible to create a `!Composed` directly specifying a
sequence of `!Composable` as arguments.
Example::
>>> comp = sql.Composed(
... [sql.SQL("insert into "), sql.Identifier("table")])
>>> print(comp.as_string(conn))
insert into "table"
`!Composed` objects are iterable (so they can be used in `SQL.join` for
instance).
"""
def __init__(self, seq):
wrapped = []
for i in seq:
if not isinstance(i, Composable):
raise TypeError(
"Composed elements must be Composable, got %r instead" % i)
wrapped.append(i)
super(Composed, self).__init__(wrapped)
@property
def seq(self):
"""The list of the content of the `!Composed`."""
return list(self._wrapped)
def as_string(self, context):
rv = []
for i in self._wrapped:
rv.append(i.as_string(context))
return ''.join(rv)
def __iter__(self):
return iter(self._wrapped)
def __add__(self, other):
if isinstance(other, Composed):
return Composed(self._wrapped + other._wrapped)
if isinstance(other, Composable):
return Composed(self._wrapped + [other])
else:
return NotImplemented
def join(self, joiner):
"""
Return a new `!Composed` interposing the *joiner* with the `!Composed` items.
The *joiner* must be a `SQL` or a string which will be interpreted as
an `SQL`.
Example::
>>> fields = sql.Identifier('foo') + sql.Identifier('bar') # a Composed
>>> print(fields.join(', ').as_string(conn))
"foo", "bar"
"""
if isinstance(joiner, string_types):
joiner = SQL(joiner)
elif not isinstance(joiner, SQL):
raise TypeError(
"Composed.join() argument must be a string or an SQL")
return joiner.join(self)
class SQL(Composable):
"""
A `Composable` representing a snippet of SQL statement.
`!SQL` exposes `join()` and `format()` methods useful to create a template
where to merge variable parts of a query (for instance field or table
names).
The *string* doesn't undergo any form of escaping, so it is not suitable to
represent variable identifiers or values: you should only use it to pass
constant strings representing templates or snippets of SQL statements; use
other objects such as `Identifier` or `Literal` to represent variable
parts.
Example::
>>> query = sql.SQL("select {0} from {1}").format(
... sql.SQL(', ').join([sql.Identifier('foo'), sql.Identifier('bar')]),
... sql.Identifier('table'))
>>> print(query.as_string(conn))
select "foo", "bar" from "table"
"""
def __init__(self, string):
if not isinstance(string, string_types):
raise TypeError("SQL values must be strings")
super(SQL, self).__init__(string)
@property
def string(self):
"""The string wrapped by the `!SQL` object."""
return self._wrapped
def as_string(self, context):
return self._wrapped
def format(self, *args, **kwargs):
"""
Merge `Composable` objects into a template.
:param `Composable` args: parameters to replace to numbered
(``{0}``, ``{1}``) or auto-numbered (``{}``) placeholders
:param `Composable` kwargs: parameters to replace to named (``{name}``)
placeholders
:return: the union of the `!SQL` string with placeholders replaced
:rtype: `Composed`
The method is similar to the Python `str.format()` method: the string
template supports auto-numbered (``{}``), numbered (``{0}``,
``{1}``...), and named placeholders (``{name}``), with positional
arguments replacing the numbered placeholders and keywords replacing
the named ones. However placeholder modifiers (``{0!r}``, ``{0:<10}``)
are not supported. Only `!Composable` objects can be passed to the
template.
Example::
>>> print(sql.SQL("select * from {} where {} = %s")
... .format(sql.Identifier('people'), sql.Identifier('id'))
... .as_string(conn))
select * from "people" where "id" = %s
>>> print(sql.SQL("select * from {tbl} where {pkey} = %s")
... .format(tbl=sql.Identifier('people'), pkey=sql.Identifier('id'))
... .as_string(conn))
select * from "people" where "id" = %s
"""
rv = []
autonum = 0
for pre, name, spec, conv in _formatter.parse(self._wrapped):
if spec:
raise ValueError("no format specification supported by SQL")
if conv:
raise ValueError("no format conversion supported by SQL")
if pre:
rv.append(SQL(pre))
if name is None:
continue
if name.isdigit():
if autonum:
raise ValueError(
"cannot switch from automatic field numbering to manual")
rv.append(args[int(name)])
autonum = None
elif not name:
if autonum is None:
raise ValueError(
"cannot switch from manual field numbering to automatic")
rv.append(args[autonum])
autonum += 1
else:
rv.append(kwargs[name])
return Composed(rv)
def join(self, seq):
"""
Join a sequence of `Composable`.
:param seq: the elements to join.
:type seq: iterable of `!Composable`
Use the `!SQL` object's *string* to separate the elements in *seq*.
Note that `Composed` objects are iterable too, so they can be used as
argument for this method.
Example::
>>> snip = sql.SQL(', ').join(
... sql.Identifier(n) for n in ['foo', 'bar', 'baz'])
>>> print(snip.as_string(conn))
"foo", "bar", "baz"
"""
rv = []
it = iter(seq)
try:
rv.append(next(it))
except StopIteration:
pass
else:
for i in it:
rv.append(self)
rv.append(i)
return Composed(rv)
class Identifier(Composable):
"""
A `Composable` representing an SQL identifier or a dot-separated sequence.
Identifiers usually represent names of database objects, such as tables or
fields. PostgreSQL identifiers follow `different rules`__ than SQL string
literals for escaping (e.g. they use double quotes instead of single).
.. __: https://www.postgresql.org/docs/current/static/sql-syntax-lexical.html# \
SQL-SYNTAX-IDENTIFIERS
Example::
>>> t1 = sql.Identifier("foo")
>>> t2 = sql.Identifier("ba'r")
>>> t3 = sql.Identifier('ba"z')
>>> print(sql.SQL(', ').join([t1, t2, t3]).as_string(conn))
"foo", "ba'r", "ba""z"
Multiple strings can be passed to the object to represent a qualified name,
i.e. a dot-separated sequence of identifiers.
Example::
>>> query = sql.SQL("select {} from {}").format(
... sql.Identifier("table", "field"),
... sql.Identifier("schema", "table"))
>>> print(query.as_string(conn))
select "table"."field" from "schema"."table"
"""
def __init__(self, *strings):
if not strings:
raise TypeError("Identifier cannot be empty")
for s in strings:
if not isinstance(s, string_types):
raise TypeError("SQL identifier parts must be strings")
super(Identifier, self).__init__(strings)
@property
def strings(self):
"""A tuple with the strings wrapped by the `Identifier`."""
return self._wrapped
@property
def string(self):
"""The string wrapped by the `Identifier`.
"""
if len(self._wrapped) == 1:
return self._wrapped[0]
else:
raise AttributeError(
"the Identifier wraps more than one than one string")
def __repr__(self):
return "%s(%s)" % (
self.__class__.__name__,
', '.join(map(repr, self._wrapped)))
def as_string(self, context):
return '.'.join(ext.quote_ident(s, context) for s in self._wrapped)
class Literal(Composable):
"""
A `Composable` representing an SQL value to include in a query.
Usually you will want to include placeholders in the query and pass values
as `~cursor.execute()` arguments. If however you really really need to
include a literal value in the query you can use this object.
The string returned by `!as_string()` follows the normal :ref:`adaptation
rules <python-types-adaptation>` for Python objects.
Example::
>>> s1 = sql.Literal("foo")
>>> s2 = sql.Literal("ba'r")
>>> s3 = sql.Literal(42)
>>> print(sql.SQL(', ').join([s1, s2, s3]).as_string(conn))
'foo', 'ba''r', 42
"""
@property
def wrapped(self):
"""The object wrapped by the `!Literal`."""
return self._wrapped
def as_string(self, context):
# is it a connection or cursor?
if isinstance(context, ext.connection):
conn = context
elif isinstance(context, ext.cursor):
conn = context.connection
else:
raise TypeError("context must be a connection or a cursor")
a = ext.adapt(self._wrapped)
if hasattr(a, 'prepare'):
a.prepare(conn)
rv = a.getquoted()
if PY3 and isinstance(rv, bytes):
rv = rv.decode(ext.encodings[conn.encoding])
return rv
class Placeholder(Composable):
"""A `Composable` representing a placeholder for query parameters.
If the name is specified, generate a named placeholder (e.g. ``%(name)s``),
otherwise generate a positional placeholder (e.g. ``%s``).
The object is useful to generate SQL queries with a variable number of
arguments.
Examples::
>>> names = ['foo', 'bar', 'baz']
>>> q1 = sql.SQL("insert into table ({}) values ({})").format(
... sql.SQL(', ').join(map(sql.Identifier, names)),
... sql.SQL(', ').join(sql.Placeholder() * len(names)))
>>> print(q1.as_string(conn))
insert into table ("foo", "bar", "baz") values (%s, %s, %s)
>>> q2 = sql.SQL("insert into table ({}) values ({})").format(
... sql.SQL(', ').join(map(sql.Identifier, names)),
... sql.SQL(', ').join(map(sql.Placeholder, names)))
>>> print(q2.as_string(conn))
insert into table ("foo", "bar", "baz") values (%(foo)s, %(bar)s, %(baz)s)
"""
def __init__(self, name=None):
if isinstance(name, string_types):
if ')' in name:
raise ValueError("invalid name: %r" % name)
elif name is not None:
raise TypeError("expected string or None as name, got %r" % name)
super(Placeholder, self).__init__(name)
@property
def name(self):
"""The name of the `!Placeholder`."""
return self._wrapped
def __repr__(self):
return "Placeholder(%r)" % (
self._wrapped if self._wrapped is not None else '',)
def as_string(self, context):
if self._wrapped is not None:
return "%%(%s)s" % self._wrapped
else:
return "%s"
# Literals
NULL = SQL("NULL")
DEFAULT = SQL("DEFAULT")
FILE:assets/driver/psycopg2/tz.py
"""tzinfo implementations for psycopg2
This module holds two different tzinfo implementations that can be used as
the 'tzinfo' argument to datetime constructors, directly passed to psycopg
functions or used to set the .tzinfo_factory attribute in cursors.
"""
# psycopg/tz.py - tzinfo implementation
#
# Copyright (C) 2003-2019 Federico Di Gregorio <[email protected]>
# Copyright (C) 2020 The Psycopg Team
#
# psycopg2 is free software: you can redistribute it and/or modify it
# under the terms of the GNU Lesser General Public License as published
# by the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# In addition, as a special exception, the copyright holders give
# permission to link this program with the OpenSSL library (or with
# modified versions of OpenSSL that use the same license as OpenSSL),
# and distribute linked combinations including the two.
#
# You must obey the GNU Lesser General Public License in all respects for
# all of the code used other than OpenSSL.
#
# psycopg2 is distributed in the hope that it will be useful, but WITHOUT
# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
# FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public
# License for more details.
import datetime
import time
ZERO = datetime.timedelta(0)
class FixedOffsetTimezone(datetime.tzinfo):
"""Fixed offset in minutes east from UTC.
This is exactly the implementation__ found in Python 2.3.x documentation,
with a small change to the `!__init__()` method to allow for pickling
and a default name in the form ``sHH:MM`` (``s`` is the sign.).
The implementation also caches instances. During creation, if a
FixedOffsetTimezone instance has previously been created with the same
offset and name that instance will be returned. This saves memory and
improves comparability.
.. __: https://docs.python.org/library/datetime.html
"""
_name = None
_offset = ZERO
_cache = {}
def __init__(self, offset=None, name=None):
if offset is not None:
self._offset = datetime.timedelta(minutes=offset)
if name is not None:
self._name = name
def __new__(cls, offset=None, name=None):
"""Return a suitable instance created earlier if it exists
"""
key = (offset, name)
try:
return cls._cache[key]
except KeyError:
tz = super(FixedOffsetTimezone, cls).__new__(cls, offset, name)
cls._cache[key] = tz
return tz
def __repr__(self):
offset_mins = self._offset.seconds // 60 + self._offset.days * 24 * 60
return "psycopg2.tz.FixedOffsetTimezone(offset=%r, name=%r)" \
% (offset_mins, self._name)
def __getinitargs__(self):
offset_mins = self._offset.seconds // 60 + self._offset.days * 24 * 60
return offset_mins, self._name
def utcoffset(self, dt):
return self._offset
def tzname(self, dt):
if self._name is not None:
return self._name
else:
seconds = self._offset.seconds + self._offset.days * 86400
hours, seconds = divmod(seconds, 3600)
minutes = seconds / 60
if minutes:
return "%+03d:%d" % (hours, minutes)
else:
return "%+03d" % hours
def dst(self, dt):
return ZERO
STDOFFSET = datetime.timedelta(seconds=-time.timezone)
if time.daylight:
DSTOFFSET = datetime.timedelta(seconds=-time.altzone)
else:
DSTOFFSET = STDOFFSET
DSTDIFF = DSTOFFSET - STDOFFSET
class LocalTimezone(datetime.tzinfo):
"""Platform idea of local timezone.
This is the exact implementation from the Python 2.3 documentation.
"""
def utcoffset(self, dt):
if self._isdst(dt):
return DSTOFFSET
else:
return STDOFFSET
def dst(self, dt):
if self._isdst(dt):
return DSTDIFF
else:
return ZERO
def tzname(self, dt):
return time.tzname[self._isdst(dt)]
def _isdst(self, dt):
tt = (dt.year, dt.month, dt.day,
dt.hour, dt.minute, dt.second,
dt.weekday(), 0, -1)
stamp = time.mktime(tt)
tt = time.localtime(stamp)
return tt.tm_isdst > 0
LOCAL = LocalTimezone()
# TO_DO: pre-generate some interesting time zones?
FILE:references/driver_info.md
# 瀚高数据库 (HighGo DB) 驱动详细说明
瀚高数据库（HighGo Database）提供的 `psycopg2` 是标准 PostgreSQL 驱动的修改版本。由于瀚高数据库支持特殊的安全特性（如国密 SM4 算法），使用原生的 `psycopg2` 可能无法连接或无法正常进行加密通信。
## 驱动目录结构 (`@psycopg2/`)
该驱动通常作为文件夹（如 `@psycopg2/`）分发，包含以下核心文件：
- `__init__.py`: 驱动入口。
- `_psycopg.so`: 编译后的 C 扩展模块。
- `libpq.so.5`: 瀚高修改版的 libpq 库，包含瀚高特有的通信逻辑。
- `libgmssl.so.3`: 国密算法支持库（可选）。
- 其他辅助模块如 `extras.py`, `sql.py` 等。
## 与标准 psycopg2 的区别
- **安全增强**: 支持 `sm3` 密码加密、`sm4` 链路加密等。
- **目录前缀**: 为了避免与环境中的 `pip install psycopg2` 冲突，瀚高通常将包名或目录名稍作修改（如 `@psycopg2`），但在代码内部仍然通过 `import psycopg2` 进行自我引用。
## 运行环境要求
- **Python 3.x**: 建议使用 3.8+。
- **操作系统**: 通常针对特定的 Linux 发行版提供编译好的 `.so` 文件。如果系统库不匹配，可能需要设置 `LD_LIBRARY_PATH` 指向该驱动目录。
## 脚本桥接机制
由于该驱动被命名为 `@psycopg2`，直接 `import` 并不被 Python 支持。`execute_query.py` 脚本通过以下方式解决：
1. 创建临时目录。
2. 将 `@psycopg2` 符号链接为 `psycopg2`。
3. 将临时目录加入 `sys.path`。
4. 动态导入 `psycopg2`。
FILE:scripts/execute_query.py
import sys
import os
import json
import argparse
import tempfile
import shutil
def execute_query(dsn, sql, driver_path=None):
# If no driver_path provided, look in the skill's assets/driver/psycopg2
if driver_path is None:
script_dir = os.path.dirname(os.path.abspath(__file__))
driver_path = os.path.join(script_dir, "..", "assets", "driver", "psycopg2")
if not os.path.exists(driver_path):
return {"status": "error", "message": "Driver not found at {}".format(driver_path)}
# Manual temporary directory for Python 2.7 compatibility
tmp_dir = tempfile.mkdtemp()
try:
target_path = os.path.join(tmp_dir, "psycopg2")
if not os.path.exists(target_path):
os.symlink(os.path.abspath(driver_path), target_path)
# Add the driver path to LD_LIBRARY_PATH for the .so libraries
env = os.environ.copy()
driver_abs_path = os.path.abspath(driver_path)
if "LD_LIBRARY_PATH" in env:
os.environ["LD_LIBRARY_PATH"] = "{}:{}".format(driver_abs_path, env['LD_LIBRARY_PATH'])
else:
os.environ["LD_LIBRARY_PATH"] = driver_abs_path
# Add the temp directory to sys.path
sys.path.insert(0, tmp_dir)
try:
import psycopg2
from psycopg2 import extras
conn = psycopg2.connect(dsn)
# Use DictCursor to get results with column names
cur = conn.cursor(cursor_factory=extras.RealDictCursor)
# Execute multiple statements if necessary
cur.execute(sql)
results = None
if cur.description:
results = cur.fetchall()
conn.commit()
cur.close()
conn.close()
return {"status": "success", "results": results}
except Exception as e:
return {"status": "error", "message": str(e)}
finally:
shutil.rmtree(tmp_dir)
if __name__ == "__main__":
parser = argparse.ArgumentParser(description="Execute SQL on HighGo DB using built-in psycopg2")
parser.add_argument("--dsn", required=True, help="Connection DSN")
parser.add_argument("--sql", required=True, help="SQL query or path to SQL file")
parser.add_argument("--driver", required=False, help="Path to psycopg2 directory (optional, uses built-in if omitted)")
args = parser.parse_args()
sql_content = args.sql
if os.path.exists(args.sql):
with open(args.sql, "r") as f:
sql_content = f.read()
result = execute_query(args.dsn, sql_content, args.driver)
print(json.dumps(result))