本文整理汇总了Python中psycopg2.extensions.register_adapter函数的典型用法代码示例。如果您正苦于以下问题:Python register_adapter函数的具体用法?Python register_adapter怎么用?Python register_adapter使用的例子?那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了register_adapter函数的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: __init__
def __init__(self, connection_string, data_quality=None):
super(PostgresConsoleDumper, self).__init__(connection_string, data_quality)
register_adapter(list, SQL_LIST)
try:
self.stdoutw = sys.stdout.buffer.write
except AttributeError:
self.stdoutw = sys.stdout.write
开发者ID:blitzr,项目名称:discogs-xml2db,代码行数:7,代码来源:postgresexporter.py
示例2: register_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
开发者ID:gopal-neosoft,项目名称:Assignment,代码行数:30,代码来源:extras.py
示例3: register_composite
def register_composite(cls, connection):
klass = cls()
db_type = klass.db_type(connection)
if db_type:
try:
cls.python_type = register_composite(
str(db_type),
connection.cursor().cursor,
globally=True,
factory=klass.factory_class()
).type
except psycopg2.ProgrammingError:
_missing_types[db_type] = cls
else:
def adapt_composite(composite):
# For safety, `composite_python_class` must have the same
# attributes as the namedtuple `python_type`'s fields, so
# that those can be escaped rather than relying on
# `__str__`.
return AsIs("(%s)::%s" % (
", ".join([
adapt(getattr(composite, field)).getquoted().decode('utf-8') for field in cls.python_type._fields
]), db_type
))
register_adapter(cls.composite_python_class, adapt_composite)
开发者ID:catmaid,项目名称:CATMAID,代码行数:26,代码来源:fields.py
示例4: register
def register():
"""Register adapters for numpy types."""
# Simple numeric types need only be converted by ``AsIs``
for numpy_type in (
numpy.int_,
numpy.intc,
numpy.intp,
numpy.int8,
numpy.int16,
numpy.int32,
numpy.int64,
numpy.uint8,
numpy.uint16,
numpy.uint32,
numpy.uint64,
numpy.float_,
numpy.float16,
numpy.float32,
numpy.float64,
):
register_adapter(numpy_type, AsIs)
# Booleans have to be converted
register_adapter(numpy.bool_, lambda v: AsIs(bool(v)))
开发者ID:pacificclimate,项目名称:modelmeta,代码行数:25,代码来源:psycopg2_adapters.py
示例5: __call__
def __call__(self, cls):
cls.REGISTERED = True
table = cls.TABLE
for attr, prop in cls.__dict__.items():
if isinstance(prop, ForeignKeyProperty):
target_table, target_name, target_model = prop.reference
REF = (table, attr, target_table, target_name, target_model)
BACKREF = (target_table, target_name, table, attr, cls)
if table in self.REFERENCES:
self.REFERENCES[table].append(REF)
else: self.REFERENCES[table] = [REF]
if target_table in self.BACKREFS:
self.BACKREFS[target_table].append(BACKREF)
else: self.BACKREFS[table] = [BACKREF]
cls.all = staticmethod(lambda: Query(self, cls))
cls._save = self.save
register_adapter(cls, ForeignKeyProperty.adapt)
if table not in self.TABLES:
self._create_table(cls)
# if the model defined also defines an index (which, by implication is also no in the database)
for field, in cls.INDEXES_DEFINED:
if isinstance(cls.__dict__[field], GeoProperty): self._create_geo_index(cls, field)
else: self._create_index(cls. field)
# if there is a model in the database that is not defined: disregard
return cls
开发者ID:ewestern,项目名称:postgorm,代码行数:26,代码来源:connection.py
示例6: register_ipaddress
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)
开发者ID:Oakafee,项目名称:Interpretation-of-New-Jersey,代码行数:27,代码来源:_ipaddress.py
示例7: register_uuid
def register_uuid(oids=None, conn_or_curs=None):
"""Create the UUID type and an uuid.UUID adapter."""
if not oids:
oid1 = 2950
oid2 = 2951
elif type(oids) == list:
oid1, oid2 = oids
else:
oid1 = oids
oid2 = 2951
def parseUUIDARRAY(data, cursor):
if data is None:
return None
elif data == "{}":
return []
else:
return [((len(x) > 0 and x != "NULL") and uuid.UUID(x) or None) for x in data[1:-1].split(",")]
_ext.UUID = _ext.new_type((oid1,), "UUID", lambda data, cursor: data and uuid.UUID(data) or None)
_ext.UUIDARRAY = _ext.new_type((oid2,), "UUID[]", parseUUIDARRAY)
_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
开发者ID:pombredanne,项目名称:Vixen,代码行数:27,代码来源:extras.py
示例8: register_psycopg2_composite
def register_psycopg2_composite(dbapi_connection, composite):
psycopg2.extras.register_composite(
composite.name,
dbapi_connection,
globally=True,
factory=composite.caster
)
def adapt_composite(value):
adapted = [
adapt(
getattr(value, column.name)
if not isinstance(column.type, TypeDecorator)
else column.type.process_bind_param(
getattr(value, column.name),
PGDialect_psycopg2()
)
)
for column in
composite.columns
]
for value in adapted:
if hasattr(value, 'prepare'):
value.prepare(dbapi_connection)
values = [
value.getquoted().decode(dbapi_connection.encoding)
if six.PY3
else value.getquoted()
for value in adapted
]
return AsIs("(%s)::%s" % (', '.join(values), composite.name))
register_adapter(composite.type_cls, adapt_composite)
开发者ID:konstantinoskostis,项目名称:sqlalchemy-utils,代码行数:33,代码来源:pg_composite.py
示例9: register_inet
def register_inet(oid=None, conn_or_curs=None):
"""Create the INET type and an Inet adapter."""
if not oid: oid = 869
_ext.INET = _ext.new_type((oid, ), "INET",
lambda data, cursor: data and Inet(data) or None)
_ext.register_type(_ext.INET, conn_or_curs)
_ext.register_adapter(Inet, lambda x: x)
return _ext.INET
开发者ID:amrik,项目名称:pyvertica,代码行数:8,代码来源:extras.py
示例10: test_no_mro_no_joy
def test_no_mro_no_joy(self):
from psycopg2.extensions import adapt, register_adapter, AsIs
class A: pass
class B(A): pass
register_adapter(A, lambda a: AsIs("a"))
self.assertRaises(psycopg2.ProgrammingError, adapt, B())
开发者ID:Southpaw-TACTIC,项目名称:Team,代码行数:8,代码来源:types_basic.py
示例11: configure
def configure(database_name, port):
from psycopg2.extras import Json
from psycopg2.extensions import register_adapter
global CONNECTION_STRING
CONNECTION_STRING = "host=localhost port={} user=postgres dbname={}".format(port, database_name)
register_adapter(dict, lambda d: Json(d))
开发者ID:treycucco,项目名称:pypgq,代码行数:9,代码来源:__init__.py
示例12: use_pendulum_for_time_types
def use_pendulum_for_time_types():
register_cast(OID_TIMESTAMP, "TIMESTAMP", cast_timestamp)
register_cast(OID_TIMESTAMPTZ, "TIMESTAMPTZ", cast_timestamptz)
register_cast(OID_DATE, "DATE", cast_date)
register_cast(OID_TIME, "TIME", cast_time)
register_cast(OID_INTERVAL, "INTERVAL", cast_interval)
register_adapter(datetime, adapt_datetime)
register_adapter(relativedelta, adapt_relativedelta)
开发者ID:djrobstep,项目名称:sqlbag,代码行数:9,代码来源:datetimes.py
示例13: __init__
def __init__(self):
self.dir = r"D:\Twitter_data_collect\*.json"
self.host = r"localhost"
self.database = "twitter"
self.user = "postgres"
self.password = "pgsql2015" # unix passwd?
self.data_dict = {}
self.list_twitts = []
register_adapter(dict, Json)
开发者ID:Fred-Wei,项目名称:twitter,代码行数:9,代码来源:twitter_dump.py
示例14: get_db_prep_value
def get_db_prep_value(self, value, connection, prepared=False):
"""Return a UUID object. Also, ensure that psycopg2 is
aware how to address that object.
"""
# Register the UUID type with psycopg2.
register_adapter(uuid.UUID, UUIDAdapter)
# Run the normal functionality.
return super(UUIDField, self).get_db_prep_value(value, connection, prepared=prepared)
开发者ID:johncarstens,项目名称:django-pgfields,代码行数:9,代码来源:uuid.py
示例15: test_adapt_most_specific
def test_adapt_most_specific(self):
from psycopg2.extensions import adapt, register_adapter, AsIs
class A(object): pass
class B(A): pass
class C(B): pass
register_adapter(A, lambda a: AsIs("a"))
register_adapter(B, lambda b: AsIs("b"))
self.assertEqual('b', adapt(C()).getquoted())
开发者ID:Southpaw-TACTIC,项目名称:Team,代码行数:10,代码来源:types_basic.py
示例16: test_no_mro_no_joy
def test_no_mro_no_joy(self):
from psycopg2.extensions import adapt, register_adapter, AsIs
class A: pass
class B(A): pass
register_adapter(A, lambda a: AsIs("a"))
try:
self.assertRaises(psycopg2.ProgrammingError, adapt, B())
finally:
del psycopg2.extensions.adapters[A, psycopg2.extensions.ISQLQuote]
开发者ID:ninoch,项目名称:Notes_Manager,代码行数:11,代码来源:test_types_basic.py
示例17: test_adapt_subtype_3
def test_adapt_subtype_3(self):
from psycopg2.extensions import adapt, register_adapter, AsIs
class A: pass
class B(A): pass
register_adapter(A, lambda a: AsIs("a"))
try:
self.assertEqual(b"a", adapt(B()).getquoted())
finally:
del psycopg2.extensions.adapters[A, psycopg2.extensions.ISQLQuote]
开发者ID:ninoch,项目名称:Notes_Manager,代码行数:11,代码来源:test_types_basic.py
示例18: register_macaddr_type
def register_macaddr_type():
from psycopg2.extensions import register_adapter, new_type, register_type, new_array_type
import psycopg2
oid = get_type_oid("NULL::macaddr")
PGTYPE = new_type((oid,), "macaddr", cast_macaddr)
register_type(PGTYPE)
register_adapter(MacAddr, adapt_macaddr)
mac_array_oid = get_type_oid("'{}'::macaddr[]")
array_of_mac = new_array_type((mac_array_oid, ), 'macaddr', psycopg2.STRING)
psycopg2.extensions.register_type(array_of_mac)
开发者ID:nwp90,项目名称:djorm-ext-pgarray,代码行数:12,代码来源:tests.py
示例19: register_hstore
def register_hstore(conn_or_curs, globally=False, unicode=False, oid=None):
"""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*
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 (it can be found 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.
.. versionchanged:: 2.4
added the *oid* parameter. If not specified, the typecaster is
installed also if |hstore| is not installed in the :sql:`public`
schema.
"""
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:
oid = oid[0] # for the moment we don't have a HSTOREARRAY
if isinstance(oid, int):
oid = (oid,)
# create and register the typecaster
if sys.version_info[0] < 3 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)
开发者ID:erochest,项目名称:sshPostGIS,代码行数:53,代码来源:extras.py
示例20: test_mro_required
def test_mro_required(self):
import psycopg2
from psycopg2.extensions import adapt, register_adapter, AsIs
# Intentionally old-style, they don't expose their MRO.
class A:
pass
class B(A):
pass
register_adapter(A, lambda a: AsIs("a"))
with self.assertRaises(psycopg2.ProgrammingError):
adapt(B())
开发者ID:vad,项目名称:psycopg2-ctypes,代码行数:13,代码来源:test_extensions.py
注:本文中的psycopg2.extensions.register_adapter函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论