本文整理汇总了Python中psycopg2.extensions.register_type函数的典型用法代码示例。如果您正苦于以下问题:Python register_type函数的具体用法?Python register_type怎么用?Python register_type使用的例子?那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了register_type函数的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: __init__
def __init__(self,dbname,user,host,password,port=None):
self.dbname = dbname
self.user = user
self.host = host
self.password = password
self.status=0
if port is None:
parameters = "dbname='"+dbname+"' user='"+user+"' host='"+host+"' password='"+password+"'"
else:
parameters = "dbname='"+dbname+"' user='"+user+"' host='"+host+"' password='"+password+"' port='"+str(port)+"'"
self.debug=''
try:
#for unicode
pg_extensions.register_type(pg_extensions.UNICODE)
#connecting to the database
self.conn = pg_connect(parameters)
#Useful for operation such as drop and insert
self.conn.set_isolation_level(0)
#enabling utf8 encode
self.conn.set_client_encoding('UNICODE')
self.cur = self.conn.cursor()
self.debug='connected to db!\t'
except:
self.debug='connection failed!\t'
self.status=1
self.conn = None
开发者ID:guthemberg,项目名称:tejo,代码行数:26,代码来源:database.py
示例2: register_inet
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
开发者ID:gopal-neosoft,项目名称:Assignment,代码行数:31,代码来源:extras.py
示例3: register_cidr
def register_cidr(oid=None, conn_or_curs=None):
"""Create the CIDR type and an Cidr adapter.
:param oid: oid for the PostgreSQL :sql:`cidr` 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.
"""
if not oid:
oid1 = 650
oid2 = 651
elif isinstance(oid, (list, tuple)):
oid1, oid2 = oid
else:
oid1 = oid
oid2 = 651
_ext.CIDR = _ext.new_type((oid1, ), "CIDR",
lambda data, cursor: data and Cidr(data) or None)
_ext.CIDRARRAY = _ext.new_array_type((oid2, ), "CIDRARRAY", _ext.CIDR)
_ext.register_type(_ext.CIDR, conn_or_curs)
_ext.register_type(_ext.CIDRARRAY, conn_or_curs)
return _ext.CIDR
开发者ID:trentclainor,项目名称:psycopg2,代码行数:26,代码来源:extras.py
示例4: 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
示例5: register_tstz_w_secs
def register_tstz_w_secs(oids=None, conn_or_curs=None):
"""Register alternate type caster for TIMESTAMP WITH TIME ZONE.
The Python datetime module cannot handle time zones with
seconds in the UTC offset. There are, however, historical
"time zones" which contain such offsets, eg. "Asia/Calcutta".
In many cases those offsets represent true local time.
If you encounter "unable to parse time" on a perfectly valid
timestamp you likely want to try this type caster. It truncates
the seconds from the time zone data and retries casting
the timestamp. Note that this will generate timestamps
which are INACCURATE by the number of seconds truncated
(unless the seconds were 00).
<oids>
which OIDs to use this type caster for,
defaults to TIMESTAMP WITH TIME ZONE
<conn_or_curs>
a cursor or connection if you want to attach
this type caster to that only, defaults to
None meaning all connections and cursors
"""
if oids is None:
oids = (1184,) # hardcoded from PostgreSQL headers
_ext.TSTZ_W_SECS = _ext.new_type(oids, "TSTZ_W_SECS", _convert_tstz_w_secs)
_ext.register_type(TSTZ_W_SECS, conn_or_curs)
return _ext.TSTZ_W_SECS
开发者ID:pombredanne,项目名称:Vixen,代码行数:30,代码来源:extras.py
示例6: test_unicode
def test_unicode(self):
cur = self.conn.cursor()
ext.register_type(ext.UNICODE, cur)
snowman = u"\u2603"
# unicode in statement
psycopg2.extras.execute_batch(cur,
"insert into testfast (id, data) values (%%s, %%s) -- %s" % snowman,
[(1, 'x')])
cur.execute("select id, data from testfast where id = 1")
self.assertEqual(cur.fetchone(), (1, 'x'))
# unicode in data
psycopg2.extras.execute_batch(cur,
"insert into testfast (id, data) values (%s, %s)",
[(2, snowman)])
cur.execute("select id, data from testfast where id = 2")
self.assertEqual(cur.fetchone(), (2, snowman))
# unicode in both
psycopg2.extras.execute_batch(cur,
"insert into testfast (id, data) values (%%s, %%s) -- %s" % snowman,
[(3, snowman)])
cur.execute("select id, data from testfast where id = 3")
self.assertEqual(cur.fetchone(), (3, snowman))
开发者ID:gencer,项目名称:psycopg2,代码行数:25,代码来源:test_fast_executemany.py
示例7: 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
示例8: register_ltree
def register_ltree(conn_or_curs, globally=False, oid=None, array_oid=None):
"""Register the ltree adapter and typecaster on the connection or cursor.
"""
register_adapter()
conn, curs, conn_or_curs = _solve_conn_curs(conn_or_curs)
if oid is None:
oid = get_oids(conn_or_curs, 'ltree')
if oid is None or not oid[0]:
raise psycopg2.ProgrammingError(
"ltree type not found in the database."
)
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
LTREE = ext.new_type(oid, "LTREE", cast_ltree)
ext.register_type(LTREE, not globally and conn_or_curs or None)
if array_oid:
LTREEARRAY = ext.new_array_type(array_oid, "LTREEARRAY", LTREE)
ext.register_type(LTREEARRAY, not globally and conn_or_curs or None)
开发者ID:dvarrazzo,项目名称:py-ltree,代码行数:33,代码来源:pg.py
示例9: register_composite
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
开发者ID:gopal-neosoft,项目名称:Assignment,代码行数:26,代码来源:extras.py
示例10: getconn
def getconn(self, create=True):
conn = pool.getconn(self.dsn)
conn.set_isolation_level(int(self.tilevel))
conn.set_client_encoding(self.encoding)
for tc in self.typecasts:
register_type(tc, conn)
return conn
开发者ID:belonesox,项目名称:dbexplorer_pgsql,代码行数:7,代码来源:db.py
示例11: register_ltree
def register_ltree(conn):
oid = _get_ltree_oids(conn)
if not oid[0]:
return False
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])
ltree = extensions.new_type(oid, "LTREE", _cast_fn)
extensions.register_type(ltree, None)
if array_oid:
ltree_array = extensions.new_array_type(array_oid, "LTREEARRAY", ltree)
extensions.register_type(ltree_array, None)
return True
开发者ID:MaxPresman,项目名称:peewee,代码行数:25,代码来源:postgres_ext.py
示例12: 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)
return _ext.INET
开发者ID:pombredanne,项目名称:Vixen,代码行数:7,代码来源:extras.py
示例13: 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
示例14: __init__
def __init__(self, parent, id):
register_type(UNICODE)
conn = psycopg2.connect(CONN_STR)
cur = conn.cursor()
cur.execute('select * from "typeOfCredit"')
cols = cur.description
row = cur.fetchall()
c = len(row)+1
wx.Frame.__init__(self, parent, id, 'Working with Types Of Credit', \
size=(720, 400), style=wx.DEFAULT_FRAME_STYLE)
self.window1 = wx.SplitterWindow(self, -1, style=wx.NO_BORDER)
self.panel1 = wx.Panel(self.window1, -1)
self.panel2 = wx.Panel(self.window1, -1)
self.Bind(wx.EVT_CLOSE, lambda event: self.Destroy())
self.download = wx.Button(self.panel2, -1, 'Download')
self.new = wx.Button(self.panel2, -1, 'New')
self.update = wx.Button(self.panel2, -1, 'Update')
self.delete = wx.Button(self.panel2, -1, 'Delete')
self.grid = wx.grid.Grid(self.panel1, -1, size=(1,1))
self.grid.CreateGrid(c, 5)
self.grid.SetRowLabelSize(40)
self.grid.SetColLabelSize(40)
self.grid.SetMinSize((500, 300))
self.grid.SetColLabelValue(0, 'id')
self.grid.SetColSize(0, 40)
self.grid.SetColLabelValue(1, 'Name')
self.grid.SetColSize(1, 150)
self.grid.SetColLabelValue(2, 'Conditions_id')
self.grid.SetColSize(2, 150)
self.grid.SetColLabelValue(3, 'Rate')
self.grid.SetColSize(3, 120)
self.grid.SetColLabelValue(4, 'Period')
self.grid.SetColSize(4,120)
self.Bind(wx.EVT_BUTTON, self.on_download, self.download)
self.Bind(wx.EVT_BUTTON, self.on_new, self.new)
self.Bind(wx.EVT_BUTTON, self.on_update, self.update)
self.Bind(wx.EVT_BUTTON, self.on_delete, self.delete)
self.grid.Bind(wx.EVT_KEY_DOWN, self.keydown)
self.panel1.SetMinSize((720, 370))
self.panel2.SetMinSize((720, 30))
self.window1.SetMinSize((720, 400))
sizer = wx.BoxSizer(wx.VERTICAL)
sizer1 = wx.BoxSizer(wx.VERTICAL)
sizer2 = wx.BoxSizer(wx.HORIZONTAL)
sizer1.Add(self.grid, -1, wx.EXPAND|wx.ADJUST_MINSIZE, 0)
sizer2.AddMany([(self.download, -1, wx.EXPAND|wx.ADJUST_MINSIZE, 0),
(self.new, -1, wx.EXPAND|wx.ADJUST_MINSIZE, 0),
(self.update, -1, wx.EXPAND|wx.ADJUST_MINSIZE, 0),
(self.delete, -1, wx.EXPAND|wx.ADJUST_MINSIZE, 0)])
self.panel1.SetSizer(sizer1)
self.panel2.SetSizer(sizer2)
self.window1.SplitHorizontally(self.panel1, self.panel2)
sizer.Add(self.window1, 1, wx.ALL|wx.EXPAND, 0)
self.SetAutoLayout(True)
self.SetSizer(sizer)
self.Layout()
self.Centre()
self.printTypes()
开发者ID:RadiotehniX,项目名称:credit_bank,代码行数:59,代码来源:TypeFrame.py
示例15: __init__
def __init__(self):
self.connection = self._get_connection_hadler()
self.curs = None
self.realcurs = None
self.rowsaffected = None
self.realcurs_arraysize = 2000
dbextensionss.register_type(dbextensionss.UNICODE)
dbextensionss.register_type(dbextensionss.UNICODEARRAY)
开发者ID:ylevinsky,项目名称:hstore,代码行数:8,代码来源:DBConnector.py
示例16: connect
def connect(cls, connectString):
args = connectString.asDict(exclude=('driver',))
args['database'] = args.pop('dbname')
conn = psycopg2.connect(**args)
conn.set_isolation_level(extensions.ISOLATION_LEVEL_AUTOCOMMIT)
extensions.register_type(extensions.UNICODE, conn)
return cls(conn)
开发者ID:pombreda,项目名称:rmake3,代码行数:8,代码来源:postgres.py
示例17: _cursor
def _cursor(self):
"""
Returns a psycopg2 DictCursor
"""
cur = self.conn.cursor(cursor_factory=DictCursor)
register_type(UNICODE, cur)
return cur
开发者ID:joealcorn,项目名称:PyPaste,代码行数:8,代码来源:__init__.py
示例18: _register_inet
def _register_inet(oid=None, conn_or_curs=None):
"""Create the INET type and an Inet adapter."""
from psycopg2 import extensions as _ext
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)
return _ext.INET
开发者ID:Cougar,项目名称:NIPAP,代码行数:8,代码来源:convert.py
示例19: register_cast
def register_cast(cls, connection):
cast_function = CAST_MAPPER[cls.type_name()]
cursor = connection.cursor()
cursor.execute(cls.sql_for_oid())
oid = cursor.description[0][1]
cursor.close()
PGTYPE = new_type((oid,), cls.type_name().upper(), cast_function)
register_type(PGTYPE)
开发者ID:berrytj,项目名称:django-orm-extensions,代码行数:9,代码来源:objects.py
示例20: __init__
def __init__(self, parent, id):
register_type(UNICODE)
conn = psycopg2.connect(CONN_STR)
cur = conn.cursor()
cur.execute('select personal_id, credit_id from credit where date_end < current_date')
cols = cur.description
row = cur.fetchall()
c = len(row)+1
wx.Frame.__init__(self, parent, id, 'Working with Database', \
size=(400, 300), style=wx.DEFAULT_FRAME_STYLE)
self.window1 = wx.SplitterWindow(self, -1, style=wx.BORDER_DEFAULT)
self.panel1 = wx.Panel(self.window1, -1)
self.panel2 = wx.Panel(self.window1, -1)
self.Bind(wx.EVT_CLOSE, lambda event: self.Destroy())
self.personal = wx.Button(self.panel2, -1, 'Personal')
self.types = wx.Button(self.panel2, -1, 'Types')
self.conditions = wx.Button(self.panel2, -1, 'Conditions')
self.credits = wx.Button(self.panel2, -1, 'Credits')
self.grid = wx.grid.Grid(self.panel1, -1, size=(1,1))
self.grid.CreateGrid(10, 2)
self.grid.SetRowLabelSize(40)
self.grid.SetColLabelSize(40)
self.grid.SetMinSize((300, 400))
self.grid.SetColLabelValue(0, 'personal_id')
self.grid.SetColSize(0, 100)
self.grid.SetColLabelValue(1, 'credit_id')
self.grid.SetColSize(1, 100)
self.Bind(wx.EVT_BUTTON, self.on_person, self.personal)
self.Bind(wx.EVT_BUTTON, self.on_type, self.types)
self.Bind(wx.EVT_BUTTON, self.on_conditions, self.conditions)
self.Bind(wx.EVT_BUTTON, self.on_credit, self.credits)
self.panel1.SetMinSize((3000, 300))
self.panel2.SetMinSize((100, 300))
self.window1.SetMinSize((520, 300))
sizer = wx.BoxSizer(wx.HORIZONTAL)
sizer1 = wx.BoxSizer(wx.HORIZONTAL)
sizer2 = wx.BoxSizer(wx.VERTICAL)
sizer1.Add(self.grid, -1, wx.EXPAND|wx.ADJUST_MINSIZE, 0)
sizer2.AddMany([(self.personal, -1, wx.EXPAND|wx.ADJUST_MINSIZE, 0),
(self.types, -1, wx.EXPAND|wx.ADJUST_MINSIZE, 0),
(self.conditions, -1, wx.EXPAND|wx.ADJUST_MINSIZE, 0),
(self.credits, -1, wx.EXPAND|wx.ADJUST_MINSIZE, 0)])
self.panel1.SetSizer(sizer1)
self.panel2.SetSizer(sizer2)
self.window1.SplitVertically(self.panel1, self.panel2)
sizer.Add(self.window1, 1, wx.ALL|wx.EXPAND, 0)
self.SetAutoLayout(True)
self.SetSizer(sizer)
self.Layout()
self.Centre()
self.printDebt()
开发者ID:RadiotehniX,项目名称:credit_bank,代码行数:56,代码来源:Program.py
注:本文中的psycopg2.extensions.register_type函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论