if isinstance(name, Composable):
name = name.as_string(conn)
- cur = conn.cursor(binary=True, row_factory=dict_row)
# This might result in a nested transaction. What we want is to leave
# the function with the connection in the state we found (either idle
# or intrans)
try:
with conn.transaction():
- cur.execute(cls._info_query, {"name": name})
+ with conn.cursor(binary=True, row_factory=dict_row) as cur:
+ cur.execute(cls._get_info_query(conn), {"name": name})
+ recs = cur.fetchall()
except e.UndefinedObject:
return None
- recs = cur.fetchall()
return cls._from_records(name, recs)
@classmethod
if isinstance(name, Composable):
name = name.as_string(conn)
- cur = conn.cursor(binary=True, row_factory=dict_row)
try:
async with conn.transaction():
- await cur.execute(cls._info_query, {"name": name})
+ async with conn.cursor(
+ binary=True, row_factory=dict_row
+ ) as cur:
+ await cur.execute(
+ cls._get_info_query(conn), {"name": name}
+ )
+ recs = await cur.fetchall()
except e.UndefinedObject:
return None
- recs = await cur.fetchall()
return cls._from_records(name, recs)
@classmethod
register_array(self, context)
- _info_query = """\
+ @classmethod
+ def _get_info_query(
+ cls, conn: "Union[Connection[Any], AsyncConnection[Any]]"
+ ) -> str:
+ return """\
SELECT
typname AS name, oid, typarray AS array_oid,
oid::regtype::text AS alt_name, typdelim AS delimiter
super().__init__(name, oid, array_oid)
self.subtype_oid = subtype_oid
- _info_query = """\
+ @classmethod
+ def _get_info_query(
+ cls, conn: "Union[Connection[Any], AsyncConnection[Any]]"
+ ) -> str:
+ return """\
SELECT t.typname AS name, t.oid AS oid, t.typarray AS array_oid,
r.rngsubtype AS subtype_oid
FROM pg_type t
"""
def _added(self, registry: "TypesRegistry") -> None:
- """Method called by the *registry* when the object is added there."""
# Map ranges subtypes to info
registry._registry[RangeInfo, self.subtype_oid] = self
+class MultirangeInfo(TypeInfo):
+ """Manage information about a multirange type."""
+
+ # TODO: expose to multirange module once added
+ # __module__ = "psycopg.types.multirange"
+
+ def __init__(
+ self,
+ name: str,
+ oid: int,
+ array_oid: int,
+ range_oid: int,
+ subtype_oid: int,
+ ):
+ super().__init__(name, oid, array_oid)
+ self.range_oid = range_oid
+ self.subtype_oid = subtype_oid
+
+ @classmethod
+ def _get_info_query(
+ cls, conn: "Union[Connection[Any], AsyncConnection[Any]]"
+ ) -> str:
+ if conn.info.server_version < 140000:
+ raise e.NotSupportedError(
+ "multirange types are only available from PostgreSQL 14"
+ )
+ return """\
+SELECT t.typname AS name, t.oid AS oid, t.typarray AS array_oid,
+ r.rngtypid AS range_oid, r.rngsubtype AS subtype_oid
+FROM pg_type t
+JOIN pg_range r ON t.oid = r.rngmultitypid
+WHERE t.oid = %(name)s::regtype
+"""
+
+ def _added(self, registry: "TypesRegistry") -> None:
+ # Map multiranges ranges and subtypes to info
+ registry._registry[MultirangeInfo, self.range_oid] = self
+ registry._registry[MultirangeInfo, self.subtype_oid] = self
+
+
class CompositeInfo(TypeInfo):
"""Manage information about a composite type."""
# Will be set by register() if the `factory` is a type
self.python_type: Optional[type] = None
- _info_query = """\
+ @classmethod
+ def _get_info_query(
+ cls, conn: "Union[Connection[Any], AsyncConnection[Any]]"
+ ) -> str:
+ return """\
SELECT
t.typname AS name, t.oid AS oid, t.typarray AS array_oid,
coalesce(a.fnames, '{}') AS field_names,
# Copyright (C) 2020-2021 The Psycopg Team
-from ._typeinfo import TypeInfo, RangeInfo, TypesRegistry
+from ._typeinfo import TypeInfo, RangeInfo, MultirangeInfo, TypesRegistry
from .abc import AdaptContext
from ._adapters_map import AdaptersMap
# Use tools/update_oids.py to update this data.
for t in [
# autogenerated: start
- # Generated from PostgreSQL 13.4
+ # Generated from PostgreSQL 14.0
TypeInfo("aclitem", 1033, 1034),
TypeInfo("bit", 1560, 1561),
TypeInfo("bool", 16, 1000, alt_name="boolean"),
RangeInfo("numrange", 3906, 3907, subtype_oid=1700),
RangeInfo("tsrange", 3908, 3909, subtype_oid=1114),
RangeInfo("tstzrange", 3910, 3911, subtype_oid=1184),
+ MultirangeInfo(
+ "datemultirange", 4535, 6155, range_oid=3912, subtype_oid=1082
+ ),
+ MultirangeInfo(
+ "int4multirange", 4451, 6150, range_oid=3904, subtype_oid=23
+ ),
+ MultirangeInfo(
+ "int8multirange", 4536, 6157, range_oid=3926, subtype_oid=20
+ ),
+ MultirangeInfo(
+ "nummultirange", 4532, 6151, range_oid=3906, subtype_oid=1700
+ ),
+ MultirangeInfo(
+ "tsmultirange", 4533, 6152, range_oid=3908, subtype_oid=1114
+ ),
+ MultirangeInfo(
+ "tstzmultirange", 4534, 6153, range_oid=3910, subtype_oid=1184
+ ),
# autogenerated: end
]:
types.add(t)
order by typname
"""
+py_multiranges_sql = """
+select
+ format('MultirangeInfo(%L, %s, %s, range_oid=%s, subtype_oid=%s),',
+ typname, oid, typarray, rngtypid, rngsubtype)
+from
+ pg_type t
+ join pg_range r on t.oid = rngmultitypid
+where
+ oid < 10000
+ and typtype = 'm'
+ and (typname !~ '^(_|pg_)' or typname = 'pg_lsn')
+order by typname
+"""
+
cython_oids_sql = """
select format('%s_OID = %s', upper(typname), oid)
from pg_type
where
oid < 10000
- and (typtype = any('{b,r}') or typname = 'record')
+ and (typtype = any('{b,r,m}') or typname = 'record')
and (typname !~ '^(_|pg_)' or typname = 'pg_lsn')
order by typname
"""
def update_python_oids() -> None:
- queries = [version_sql, py_types_sql, py_ranges_sql]
+ queries = [version_sql, py_types_sql, py_ranges_sql, py_multiranges_sql]
fn = ROOT / "psycopg/psycopg/postgres.py"
update_file(fn, queries)
sp.check_call(["black", "-q", fn])
projects / thirdparty / psycopg.git / commitdiff
