psycopg/psycopg2
1
1
Fork 0
mirror of https://github.com/psycopg/psycopg2.git synced 2024-11-10 19:16:34 +03:00
Code Issues Packages Projects Releases Wiki Activity

Dropped examples dir (and some leftover reference to the sandbox dir)

Browse Source 
Close #645.
This commit is contained in:
Daniele Varrazzo 2018-11-16 17:06:06 +00:00
parent f9d6430ae4
commit 654be4784c
25 changed files with 4 additions and 1812 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
MANIFEST.in
View File

@ -1,7 +1,6 @@
recursive-include psycopg *.c *.h *.manifest
recursive-include lib *.py
recursive-include tests *.py
recursive-include examples *.py somehackers.jpg whereareyou.jpg
include doc/README.rst doc/SUCCESS doc/COPYING.LESSER doc/pep-0249.txt
include doc/Makefile doc/requirements.txt
recursive-include doc/src *.rst *.py *.css Makefile

7
NEWS
View File

@ -1012,7 +1012,7 @@ What's new in psycopg 2.0 beta 7
What's new in psycopg 2.0 beta 6
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
* Support for named cursors (see examples/fetch.py).
* Support for named cursors.
* Safer parsing of time intervals.
@ -1042,7 +1042,7 @@ What's new in psycopg 2.0 beta 5
* All classes have been renamed to exist in the psycopg2._psycopg module,
to fix problems with automatic documentation generators like epydoc.
* NOTIFY is correctly trapped (see examples/notify.py for example code.)
* NOTIFY is correctly trapped.
What's new in psycopg 2.0 beta 4
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
@ -1159,8 +1159,7 @@ What's new in psycopg 1.99.10
What's new in psycopg 1.99.9
^^^^^^^^^^^^^^^^^^^^^^^^^^^^
* Added simple pooling code (psycopg.pool module); see the reworked
examples/threads.py for example code.
* Added simple pooling code (psycopg.pool module).
* Added DECIMAL typecaster to convert postgresql DECIMAL and NUMERIC
types (i.e, all types with an OID of NUMERICOID.) Note that the

90
examples/binary.py
View File

@ -1,90 +0,0 @@
# binary.py - working with binary data
#
# Copyright (C) 2001-2010 Federico Di Gregorio <fog@debian.org>
#
# 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.
#
# 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.
## put in DSN your DSN string
DSN = 'dbname=test'
## don't modify anything below this line (except for experimenting)
from __future__ import print_function
import sys
import psycopg2
if len(sys.argv) > 1:
DSN = sys.argv[1]
print("Opening connection using dsn:", DSN)
conn = psycopg2.connect(DSN)
print("Encoding for this connection is", conn.encoding)
curs = conn.cursor()
try:
curs.execute("CREATE TABLE test_binary (id int4, name text, img bytea)")
except:
conn.rollback()
curs.execute("DROP TABLE test_binary")
curs.execute("CREATE TABLE test_binary (id int4, name text, img bytea)")
conn.commit()
# first we try two inserts, one with an explicit Binary call and the other
# using a buffer on a file object.
data1 = {'id':1, 'name':'somehackers.jpg',
'img':psycopg2.Binary(open('somehackers.jpg').read())}
data2 = {'id':2, 'name':'whereareyou.jpg',
'img':buffer(open('whereareyou.jpg').read())}
curs.execute("""INSERT INTO test_binary
VALUES (%(id)s, %(name)s, %(img)s)""", data1)
curs.execute("""INSERT INTO test_binary
VALUES (%(id)s, %(name)s, %(img)s)""", data2)
# now we try to extract the images as simple text strings
print("Extracting the images as strings...")
curs.execute("SELECT * FROM test_binary")
for row in curs.fetchall():
name, ext = row[1].split('.')
new_name = name + '_S.' + ext
print(" writing %s to %s ..." % (name+'.'+ext, new_name), end=' ')
open(new_name, 'wb').write(row[2])
print("done")
print(" python type of image data is", type(row[2]))
# extract exactly the same data but using a binary cursor
print("Extracting the images using a binary cursor:")
curs.execute("""DECLARE zot CURSOR FOR
SELECT img, name FROM test_binary FOR READ ONLY""")
curs.execute("""FETCH ALL FROM zot""")
for row in curs.fetchall():
name, ext = row[1].split('.')
new_name = name + '_B.' + ext
print(" writing %s to %s ..." % (name+'.'+ext, new_name), end=' ')
open(new_name, 'wb').write(row[0])
print("done")
print(" python type of image data is", type(row[0]))
# this rollback is required because we can't drop a table with a binary cursor
# declared and still open
conn.rollback()
curs.execute("DROP TABLE test_binary")
conn.commit()
print("\nNow try to load the new images, to check it worked!")

174
examples/copy_from.py
View File

@ -1,174 +0,0 @@
# copy_from.py -- example about copy_from
#
# Copyright (C) 2002 Tom Jenkins <tjenkins@devis.com>
# Copyright (C) 2005 Federico Di Gregorio <fog@initd.org>
#
# 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.
#
# 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.
## put in DSN your DSN string
DSN = 'dbname=test'
## don't modify anything below this line (except for experimenting)
import sys
import os
import StringIO
import psycopg2
if len(sys.argv) > 1:
DSN = sys.argv[1]
print("Opening connection using dsn:", DSN)
conn = psycopg2.connect(DSN)
print("Encoding for this connection is", conn.encoding)
curs = conn.cursor()
try:
curs.execute("CREATE TABLE test_copy (fld1 text, fld2 text, fld3 int4)")
except:
conn.rollback()
curs.execute("DROP TABLE test_copy")
curs.execute("CREATE TABLE test_copy (fld1 text, fld2 text, fld3 int4)")
conn.commit()
# copy_from with default arguments, from open file
io = open('copy_from.txt', 'wr')
data = ['Tom\tJenkins\t37\n',
'Madonna\t\\N\t45\n',
'Federico\tDi Gregorio\t\\N\n']
io.writelines(data)
io.close()
io = open('copy_from.txt', 'r')
curs.copy_from(io, 'test_copy')
print("1) Copy %d records from file object " % len(data) +
"using defaults (sep: \\t and null = \\N)")
io.close()
curs.execute("SELECT * FROM test_copy")
rows = curs.fetchall()
print(" Select returned %d rows" % len(rows))
for r in rows:
print(" %s %s\t%s" % (r[0], r[1], r[2]))
curs.execute("delete from test_copy")
conn.commit()
# copy_from using custom separator, from open file
io = open('copy_from.txt', 'wr')
data = ['Tom:Jenkins:37\n',
'Madonna:\N:45\n',
'Federico:Di Gregorio:\N\n']
io.writelines(data)
io.close()
io = open('copy_from.txt', 'r')
curs.copy_from(io, 'test_copy', ':')
print("2) Copy %d records from file object using sep = :" % len(data))
io.close()
curs.execute("SELECT * FROM test_copy")
rows = curs.fetchall()
print(" Select returned %d rows" % len(rows))
for r in rows:
print(" %s %s\t%s" % (r[0], r[1], r[2]))
curs.execute("delete from test_copy")
conn.commit()
# copy_from using custom null identifier, from open file
io = open('copy_from.txt', 'wr')
data = ['Tom\tJenkins\t37\n',
'Madonna\tNULL\t45\n',
'Federico\tDi Gregorio\tNULL\n']
io.writelines(data)
io.close()
io = open('copy_from.txt', 'r')
curs.copy_from(io, 'test_copy', null='NULL')
print("3) Copy %d records from file object using null = NULL" % len(data))
io.close()
curs.execute("SELECT * FROM test_copy")
rows = curs.fetchall()
print(" Select using cursor returned %d rows" % len(rows))
for r in rows:
print(" %s %s\t%s" % (r[0], r[1], r[2]))
curs.execute("delete from test_copy")
conn.commit()
# copy_from using custom separator and null identifier
io = open('copy_from.txt', 'wr')
data = ['Tom:Jenkins:37\n', 'Madonna:NULL:45\n', 'Federico:Di Gregorio:NULL\n']
io.writelines(data)
io.close()
io = open('copy_from.txt', 'r')
curs.copy_from(io, 'test_copy', ':', 'NULL')
print("4) Copy %d records from file object " % len(data) +
"using sep = : and null = NULL")
io.close()
curs.execute("SELECT * FROM test_copy")
rows = curs.fetchall()
print(" Select using cursor returned %d rows" % len(rows))
for r in rows:
print(" %s %s\t%s" % (r[0], r[1], r[2]))
curs.execute("delete from test_copy")
conn.commit()
# anything can be used as a file if it has .read() and .readline() methods
data = StringIO.StringIO()
data.write('\n'.join(['Tom\tJenkins\t37',
'Madonna\t\N\t45',
'Federico\tDi Gregorio\t\N']))
data.seek(0)
curs.copy_from(data, 'test_copy')
print("5) Copy 3 records from StringIO object using defaults")
curs.execute("SELECT * FROM test_copy")
rows = curs.fetchall()
print(" Select using cursor returned %d rows" % len(rows))
for r in rows:
print(" %s %s\t%s" % (r[0], r[1], r[2]))
curs.execute("delete from test_copy")
conn.commit()
# simple error test
print("6) About to raise an error")
data = StringIO.StringIO()
data.write('\n'.join(['Tom\tJenkins\t37',
'Madonna\t\N\t45',
'Federico\tDi Gregorio\taaa']))
data.seek(0)
try:
curs.copy_from(data, 'test_copy')
except StandardError as err:
conn.rollback()
print(" Caught error (as expected):\n", err)
conn.rollback()
curs.execute("DROP TABLE test_copy")
os.unlink('copy_from.txt')
conn.commit()

104
examples/copy_to.py
View File

@ -1,104 +0,0 @@
# copy_to.py -- example about copy_to
#
# Copyright (C) 2002 Tom Jenkins <tjenkins@devis.com>
# Copyright (C) 2005 Federico Di Gregorio <fog@initd.org>
#
# 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.
#
# 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.
## put in DSN your DSN string
DSN = 'dbname=test'
## don't modify anything below this line (except for experimenting)
from __future__ import print_function
import sys
import os
import StringIO
import psycopg2
if len(sys.argv) > 1:
DSN = sys.argv[1]
print("Opening connection using dsn:", DSN)
conn = psycopg2.connect(DSN)
print("Encoding for this connection is", conn.encoding)
curs = conn.cursor()
try:
curs.execute("CREATE TABLE test_copy (fld1 text, fld2 text, fld3 int4)")
except:
conn.rollback()
curs.execute("DROP TABLE test_copy")
curs.execute("CREATE TABLE test_copy (fld1 text, fld2 text, fld3 int4)")
conn.commit()
# demostrate copy_to functionality
data = [('Tom', 'Jenkins', '37'),
('Madonna', None, '45'),
('Federico', 'Di Gregorio', None)]
query = "INSERT INTO test_copy VALUES (%s, %s, %s)"
curs.executemany(query, data)
conn.commit()
# copy_to using defaults
io = open('copy_to.txt', 'w')
curs.copy_to(io, 'test_copy')
print("1) Copy %d records into file object using defaults: " % len (data) + \
"sep = \\t and null = \\N")
io.close()
rows = open('copy_to.txt', 'r').readlines()
print(" File has %d rows:" % len(rows))
for r in rows:
print(" ", r, end=' ')
# copy_to using custom separator
io = open('copy_to.txt', 'w')
curs.copy_to(io, 'test_copy', ':')
print("2) Copy %d records into file object using sep = :" % len(data))
io.close()
rows = open('copy_to.txt', 'r').readlines()
print(" File has %d rows:" % len(rows))
for r in rows:
print(" ", r, end=' ')
# copy_to using custom null identifier
io = open('copy_to.txt', 'w')
curs.copy_to(io, 'test_copy', null='NULL')
print("3) Copy %d records into file object using null = NULL" % len(data))
io.close()
rows = open('copy_to.txt', 'r').readlines()
print(" File has %d rows:" % len(rows))
for r in rows:
print(" ", r, end=' ')
# copy_to using custom separator and null identifier
io = open('copy_to.txt', 'w')
curs.copy_to(io, 'test_copy', ':', 'NULL')
print("4) Copy %d records into file object using sep = : and null ) NULL" % \
len(data))
io.close()
rows = open('copy_to.txt', 'r').readlines()
print(" File has %d rows:" % len(rows))
for r in rows:
print(" ", r, end=' ')
curs.execute("DROP TABLE test_copy")
os.unlink('copy_to.txt')
conn.commit()

63
examples/cursor.py
View File

@ -1,63 +0,0 @@
# cursor.py - how to subclass the cursor type
#
# Copyright (C) 2004-2010 Federico Di Gregorio <fog@debian.org>
#
# 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.
#
# 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.
## put in DSN your DSN string
DSN = 'dbname=test'
## don't modify anything below this line (except for experimenting)
import sys
import psycopg2
import psycopg2.extensions
if len(sys.argv) > 1:
DSN = sys.argv[1]
print("Opening connection using dsn:", DSN)
conn = psycopg2.connect(DSN)
print("Encoding for this connection is", conn.encoding)
class NoDataError(psycopg2.ProgrammingError):
"""Exception that will be raised by our cursor."""
pass
class Cursor(psycopg2.extensions.cursor):
"""A custom cursor."""
def fetchone(self):
"""Like fetchone but raise an exception if no data is available.
Note that to have .fetchmany() and .fetchall() to raise the same
exception we'll have to override them too; even if internally psycopg
uses the same function to fetch rows, the code path from Python is
different.
"""
d = psycopg2.extensions.cursor.fetchone(self)
if d is None:
raise NoDataError("no more data")
return d
curs = conn.cursor(cursor_factory=Cursor)
curs.execute("SELECT 1 AS foo")
print("Result of fetchone():", curs.fetchone())
# now let's raise the exception
try:
curs.fetchone()
except NoDataError as err:
print("Exception caught:", err)
conn.rollback()

137
examples/dialtone.py
View File

@ -1,137 +0,0 @@
"""
This example/recipe has been contributed by Valentino Volonghi (dialtone)
Mapping arbitrary objects to a PostgreSQL database with psycopg2
- Problem
You need to store arbitrary objects in a PostgreSQL database without being
intrusive for your classes (don't want inheritance from an 'Item' or
'Persistent' object).
- Solution
"""
from datetime import datetime
import psycopg2
from psycopg2.extensions import adapt, register_adapter
# Here is the adapter for every object that we may ever need to
# insert in the database. It receives the original object and does
# its job on that instance
class ObjectMapper(object):
def __init__(self, orig, curs=None):
self.orig = orig
self.tmp = {}
self.items, self.fields = self._gatherState()
def _gatherState(self):
adaptee_name = self.orig.__class__.__name__
fields = sorted([(field, getattr(self.orig, field))
for field in persistent_fields[adaptee_name]])
items = []
for item, value in fields:
items.append(item)
return items, fields
def getTableName(self):
return self.orig.__class__.__name__
def getMappedValues(self):
tmp = []
for i in self.items:
tmp.append("%%(%s)s"%i)
return ", ".join(tmp)
def getValuesDict(self):
return dict(self.fields)
def getFields(self):
return self.items
def generateInsert(self):
qry = "INSERT INTO"
qry += " " + self.getTableName() + " ("
qry += ", ".join(self.getFields()) + ") VALUES ("
qry += self.getMappedValues() + ")"
return qry, self.getValuesDict()
# Here are the objects
class Album(object):
id = 0
def __init__(self):
self.creation_time = datetime.now()
self.album_id = self.id
Album.id = Album.id + 1
self.binary_data = buffer('12312312312121')
class Order(object):
id = 0
def __init__(self):
self.items = ['rice','chocolate']
self.price = 34
self.order_id = self.id
Order.id = Order.id + 1
register_adapter(Album, ObjectMapper)
register_adapter(Order, ObjectMapper)
# Describe what is needed to save on each object
# This is actually just configuration, you can use xml with a parser if you
# like to have plenty of wasted CPU cycles ;P.
persistent_fields = {'Album': ['album_id', 'creation_time', 'binary_data'],
'Order': ['order_id', 'items', 'price']
}
print(adapt(Album()).generateInsert())
print(adapt(Album()).generateInsert())
print(adapt(Album()).generateInsert())
print(adapt(Order()).generateInsert())
print(adapt(Order()).generateInsert())
print(adapt(Order()).generateInsert())
"""
- Discussion
Psycopg 2 has a great new feature: adaptation. The big thing about
adaptation is that it enables the programmer to glue most of the
code out there without many difficulties.
This recipe tries to focus attention on a way to generate SQL queries to
insert completely new objects inside a database. As you can see objects do
not know anything about the code that is handling them. We specify all the
fields that we need for each object through the persistent_fields dict.
The most important lines of this recipe are:
register_adapter(Album, ObjectMapper)
register_adapter(Order, ObjectMapper)
In these lines we notify the system that when we call adapt with an Album instance
as an argument we want it to istantiate ObjectMapper passing the Album instance
as argument (self.orig in the ObjectMapper class).
The output is something like this (for each call to generateInsert):
('INSERT INTO Album (album_id, binary_data, creation_time) VALUES
(%(album_id)s, %(binary_data)s, %(creation_time)s)',
{'binary_data': <read-only buffer for 0x402de070, ...>,
'creation_time': datetime.datetime(2004, 9, 10, 20, 48, 29, 633728),
'album_id': 1}
)
This is a tuple of {SQL_QUERY, FILLING_DICT}, and all the quoting/converting
stuff (from python's datetime to postgres s and from python's buffer to
postgres' blob) is handled with the same adaptation process hunder the hood
by psycopg2.
At last, just notice that ObjectMapper is working for both Album and Order
instances without any glitches at all, and both classes could have easily been
coming from closed source libraries or C coded ones (which are not easily
modified), whereas a common pattern in todays ORMs or OODBs is to provide
a basic 'Persistent' object that already knows how to store itself in the
database.
"""

65
examples/dict.py
View File

@ -1,65 +0,0 @@
# dict.py - using DictCUrsor/DictRow
#
# Copyright (C) 2005-2010 Federico Di Gregorio <fog@debian.org>
#
# 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.
#
# 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.
## put in DSN your DSN string
DSN = 'dbname=test'
## don't modify anything below this line (except for experimenting)
import sys
import psycopg2
import psycopg2.extras
if len(sys.argv) > 1:
DSN = sys.argv[1]
print("Opening connection using dsn:", DSN)
conn = psycopg2.connect(DSN)
print("Encoding for this connection is", conn.encoding)
curs = conn.cursor(cursor_factory=psycopg2.extras.DictCursor)
curs.execute("SELECT 1 AS foo, 'cip' AS bar, date(now()) as zot")
print("Cursor's row factory is", curs.row_factory)
data = curs.fetchone()
print("The type of the data row is", type(data))
print("Some data accessed both as tuple and dict:")
print(" ", data['foo'], data['bar'], data['zot'])
print(" ", data[0], data[1], data[2])
# execute another query and demostrate we can still access the row
curs.execute("SELECT 2 AS foo")
print("The type of the data row is", type(data))
print("Some more data accessed both as tuple and dict:")
print(" ", data['foo'], data['bar'], data['zot'])
print(" ", data[0], data[1], data[2])
curs = conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor)
curs.execute("SELECT 1 AS foo, 'cip' AS bar, date(now()) as zot")
print("Cursor's row factory is", curs.row_factory)
data = curs.fetchone()
print("The type of the data row is", type(data))
print("Some data accessed both as tuple and dict:")
print(" ", data['foo'], data['bar'], data['zot'])
print(" ", "No access using indices: this is a specialized cursor.")
# execute another query and demostrate we can still access the row
curs.execute("SELECT 2 AS foo")
print("The type of the data row is", type(data))
print("Some more data accessed both as tuple and dict:")
print(" ", data['foo'], data['bar'], data['zot'])
print(" ", "No access using indices: this is a specialized cursor.")

99
examples/dt.py
View File

@ -1,99 +0,0 @@
# datetime.py - example of using date and time types
#
# Copyright (C) 2001-2010 Federico Di Gregorio <fog@debian.org>
#
# 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.
#
# 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.
## put in DSN your DSN string
DSN = 'dbname=test'
## don't modify anything below this line (except for experimenting)
import sys
import psycopg2
import mx.DateTime
import datetime
from psycopg2.extensions import adapt
if len(sys.argv) > 1:
DSN = sys.argv[1]
print("Opening connection using dsn:", DSN)
conn = psycopg2.connect(DSN)
curs = conn.cursor()
try:
curs.execute("""CREATE TABLE test_dt (
k int4, d date, t time, dt timestamp, z interval)""")
except:
conn.rollback()
curs.execute("DROP TABLE test_dt")
curs.execute("""CREATE TABLE test_dt (
k int4, d date, t time, dt timestamp, z interval)""")
conn.commit()
# build and insert some data using mx.DateTime
mx1 = (
1,
mx.DateTime.Date(2004, 10, 19),
mx.DateTime.Time(0, 11, 17.015),
mx.DateTime.Timestamp(2004, 10, 19, 0, 11, 17.5),
mx.DateTime.DateTimeDelta(13, 15, 17, 59.9))
from psycopg2.extensions import adapt
import psycopg2.extras
print(adapt(mx1))
print("Inserting mx.DateTime values...")
curs.execute("INSERT INTO test_dt VALUES (%s, %s, %s, %s, %s)", mx1)
# build and insert some values using the datetime adapters
dt1 = (
2,
datetime.date(2004, 10, 19),
datetime.time(0, 11, 17, 15000),
datetime.datetime(2004, 10, 19, 0, 11, 17, 500000),
datetime.timedelta(13, 15*3600+17*60+59, 900000))
print("Inserting Python datetime values...")
curs.execute("INSERT INTO test_dt VALUES (%s, %s, %s, %s, %s)", dt1)
# now extract the row from database and print them
print("Extracting values inserted with mx.DateTime wrappers:")
curs.execute("SELECT d, t, dt, z FROM test_dt WHERE k = 1")
for n, x in zip(mx1[1:], curs.fetchone()):
try:
# this will work only if psycopg has been compiled with datetime
# as the default typecaster for date/time values
s = repr(n) + "\n -> " + str(adapt(n)) + \
"\n -> " + repr(x) + "\n -> " + x.isoformat()
except:
s = repr(n) + "\n -> " + str(adapt(n)) + \
"\n -> " + repr(x) + "\n -> " + str(x)
print(s)
print()
print("Extracting values inserted with Python datetime wrappers:")
curs.execute("SELECT d, t, dt, z FROM test_dt WHERE k = 2")
for n, x in zip(dt1[1:], curs.fetchone()):
try:
# this will work only if psycopg has been compiled with datetime
# as the default typecaster for date/time values
s = repr(n) + "\n -> " + repr(x) + "\n -> " + x.isoformat()
except:
s = repr(n) + "\n -> " + repr(x) + "\n -> " + str(x)
print(s)
print()
curs.execute("DROP TABLE test_dt")
conn.commit()

105
examples/encoding.py
View File

@ -1,105 +0,0 @@
# encoding.py - show to change client encoding (and test it works)
# -*- encoding: utf8 -*-
#
# Copyright (C) 2004-2010 Federico Di Gregorio <fog@debian.org>
#
# 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.
#
# 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.
## put in DSN your DSN string
DSN = 'dbname=test'
## don't modify anything below this line (except for experimenting)
import sys
import psycopg2
import psycopg2.extensions
if len(sys.argv) > 1:
DSN = sys.argv[1]
print("Opening connection using dsn:", DSN)
conn = psycopg2.connect(DSN)
print("Initial encoding for this connection is", conn.encoding)
print("\n** This example is supposed to be run in a UNICODE terminal! **\n")
print("Available encodings:")
encs = psycopg2.extensions.encodings.items()
encs.sort()
for a, b in encs:
print(" ", a, "<->", b)
print("Using STRING typecaster")
print("Setting backend encoding to LATIN1 and executing queries:")
conn.set_client_encoding('LATIN1')
curs = conn.cursor()
curs.execute("SELECT %s::TEXT AS foo", ('àèìòù',))
x = curs.fetchone()[0]
print(" ->", unicode(x, 'latin-1').encode('utf-8'), type(x))
curs.execute("SELECT %s::TEXT AS foo", (u'àèìòù',))
x = curs.fetchone()[0]
print(" ->", unicode(x, 'latin-1').encode('utf-8'), type(x))
print("Setting backend encoding to UTF8 and executing queries:")
conn.set_client_encoding('UNICODE')
curs = conn.cursor()
curs.execute("SELECT %s::TEXT AS foo", (u'àèìòù'.encode('utf-8'),))
x = curs.fetchone()[0]
print(" ->", x, type(x))
curs.execute("SELECT %s::TEXT AS foo", (u'àèìòù',))
x = curs.fetchone()[0]
print(" ->", x, type(x))
print("Using UNICODE typecaster")
psycopg2.extensions.register_type(psycopg2.extensions.UNICODE)
print("Setting backend encoding to LATIN1 and executing queries:")
conn.set_client_encoding('LATIN1')
curs = conn.cursor()
curs.execute("SELECT %s::TEXT AS foo", ('àèìòù',))
x = curs.fetchone()[0]
print(" ->", x.encode('utf-8'), ":", type(x))
curs.execute("SELECT %s::TEXT AS foo", (u'àèìòù',))
x = curs.fetchone()[0]
print(" ->", x.encode('utf-8'), ":", type(x))
print("Setting backend encoding to UTF8 and executing queries:")
conn.set_client_encoding('UNICODE')
curs = conn.cursor()
curs.execute("SELECT %s::TEXT AS foo", (u'àèìòù'.encode('utf-8'),))
x = curs.fetchone()[0]
print(" ->", x.encode('utf-8'), ":", type(x))
curs.execute("SELECT %s::TEXT AS foo", (u'àèìòù',))
x = curs.fetchone()[0]
print(" ->", x.encode('utf-8'), ":", type(x))
print("Executing full UNICODE queries")
print("Setting backend encoding to LATIN1 and executing queries:")
conn.set_client_encoding('LATIN1')
curs = conn.cursor()
curs.execute(u"SELECT %s::TEXT AS foo", ('àèìòù',))
x = curs.fetchone()[0]
print(" ->", x.encode('utf-8'), ":", type(x))
curs.execute(u"SELECT %s::TEXT AS foo", (u'àèìòù',))
x = curs.fetchone()[0]
print(" ->", x.encode('utf-8'), ":", type(x))
print("Setting backend encoding to UTF8 and executing queries:")
conn.set_client_encoding('UNICODE')
curs = conn.cursor()
curs.execute(u"SELECT %s::TEXT AS foo", (u'àèìòù'.encode('utf-8'),))
x = curs.fetchone()[0]
print(" ->", x.encode('utf-8'), ":", type(x))
curs.execute(u"SELECT %s::TEXT AS foo", (u'àèìòù',))
x = curs.fetchone()[0]
print(" ->", x.encode('utf-8'), ":", type(x))

80
examples/fetch.py
View File

@ -1,80 +0,0 @@
# fetch.py -- example about declaring cursors
#
# Copyright (C) 2001-2010 Federico Di Gregorio <fog@debian.org>
#
# 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.
#
# 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.
## put in DSN your DSN string
DSN = 'dbname=test'
## don't modify anything below this line (except for experimenting)
import sys
import psycopg2
if len(sys.argv) > 1:
DSN = sys.argv[1]
print("Opening connection using dsn:", DSN)
conn = psycopg2.connect(DSN)
print("Encoding for this connection is", conn.encoding)
curs = conn.cursor()
try:
curs.execute("CREATE TABLE test_fetch (val int4)")
except:
conn.rollback()
curs.execute("DROP TABLE test_fetch")
curs.execute("CREATE TABLE test_fetch (val int4)")
conn.commit()
# we use this function to format the output
def flatten(l):
"""Flattens list of tuples l."""
return map(lambda x: x[0], l)
# insert 20 rows in the table
for i in range(20):
curs.execute("INSERT INTO test_fetch VALUES(%s)", (i,))
conn.commit()
# does some nice tricks with the transaction and postgres cursors
# (remember to always commit or rollback before a DECLARE)
#
# we don't need to DECLARE ourselves, psycopg now supports named
# cursors (but we leave the code here, comments, as an example of
# what psycopg is doing under the hood)
#
#curs.execute("DECLARE crs CURSOR FOR SELECT * FROM test_fetch")
#curs.execute("FETCH 10 FROM crs")
#print "First 10 rows:", flatten(curs.fetchall())
#curs.execute("MOVE -5 FROM crs")
#print "Moved back cursor by 5 rows (to row 5.)"
#curs.execute("FETCH 10 FROM crs")
#print "Another 10 rows:", flatten(curs.fetchall())
#curs.execute("FETCH 10 FROM crs")
#print "The remaining rows:", flatten(curs.fetchall())
ncurs = conn.cursor("crs")
ncurs.execute("SELECT * FROM test_fetch")
print("First 10 rows:", flatten(ncurs.fetchmany(10)))
ncurs.scroll(-5)
print("Moved back cursor by 5 rows (to row 5.)")
print("Another 10 rows:", flatten(ncurs.fetchmany(10)))
print("Another one:", list(ncurs.fetchone()))
print("The remaining rows:", flatten(ncurs.fetchall()))
conn.rollback()
curs.execute("DROP TABLE test_fetch")
conn.commit()

59
examples/lastrowid.py
View File

@ -1,59 +0,0 @@
# lastrowid.py - example of using .lastrowid attribute
#
# Copyright (C) 2001-2010 Federico Di Gregorio <fog@debian.org>
#
# 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.
#
# 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.
## put in DSN your DSN string
DSN = 'dbname=test'
## don't modify anything below this line (except for experimenting)
import sys, psycopg2
if len(sys.argv) > 1:
DSN = sys.argv[1]
print("Opening connection using dsn:", DSN)
conn = psycopg2.connect(DSN)
curs = conn.cursor()
try:
curs.execute("CREATE TABLE test_oid (name text, surname text)")
except:
conn.rollback()
curs.execute("DROP TABLE test_oid")
curs.execute("CREATE TABLE test_oid (name text, surname text)")
conn.commit()
data = ({'name':'Federico', 'surname':'Di Gregorio'},
{'name':'Pierluigi', 'surname':'Di Nunzio'})
curs.execute("""INSERT INTO test_oid
VALUES (%(name)s, %(surname)s)""", data[0])
foid = curs.lastrowid
print("Oid for %(name)s %(surname)s" % data[0], "is", foid)
curs.execute("""INSERT INTO test_oid
VALUES (%(name)s, %(surname)s)""", data[1])
moid = curs.lastrowid
print("Oid for %(name)s %(surname)s" % data[1], "is", moid)
curs.execute("SELECT * FROM test_oid WHERE oid = %s", (foid,))
print("Oid", foid, "selected %s %s" % curs.fetchone())
curs.execute("SELECT * FROM test_oid WHERE oid = %s", (moid,))
print("Oid", moid, "selected %s %s" % curs.fetchone())
curs.execute("DROP TABLE test_oid")
conn.commit()

91
examples/lobject.py
View File

@ -1,91 +0,0 @@
# lobject.py - lobject example
#
# Copyright (C) 2001-2006 Federico Di Gregorio <fog@debian.org>
#
# 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.
#
# 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.
## put in DSN your DSN string
DSN = 'dbname=test'
## don't modify anything below this line (except for experimenting)
import sys
import psycopg2
if len(sys.argv) > 1:
DSN = sys.argv[1]
print("Opening connection using dsn:", DSN)
conn = psycopg2.connect(DSN)
print("Encoding for this connection is", conn.encoding)
# this will create a large object with a new random oid, we'll
# use it to make some basic tests about read/write and seek.
lobj = conn.lobject()
loid = lobj.oid
print("Created a new large object with oid", loid)
print("Manually importing some binary data into the object:")
data = open("somehackers.jpg").read()
len = lobj.write(data)
print(" imported", len, "bytes of data")
conn.commit()
print("Trying to (re)open large object with oid", loid)
lobj = conn.lobject(loid)
print("Manually exporting the data from the lobject:")
data1 = lobj.read()
len = lobj.tell()
lobj.seek(0, 0)
data2 = lobj.read()
if data1 != data2:
print("ERROR: read after seek returned different data")
open("somehackers_lobject1.jpg", 'wb').write(data1)
print(" written", len, "bytes of data to somehackers_lobject1.jpg")
lobj.unlink()
print("Large object with oid", loid, "removed")
conn.commit()
# now we try to use the import and export functions to do the same
lobj = conn.lobject(0, 'n', 0, "somehackers.jpg")
loid = lobj.oid
print("Imported a new large object with oid", loid)
conn.commit()
print("Trying to (re)open large object with oid", loid)
lobj = conn.lobject(loid, 'n')
print("Using export() to export the data from the large object:")
lobj.export("somehackers_lobject2.jpg")
print(" exported large object to somehackers_lobject2.jpg")
lobj.unlink()
print("Large object with oid", loid, "removed")
conn.commit()
# this will create a very large object with a new random oid.
lobj = conn.lobject()
loid = lobj.oid
print("Created a new large object with oid", loid)
print("Manually importing a lot of data into the object:")
data = "data" * 1000000
len = lobj.write(data)
print(" imported", len, "bytes of data")
conn.rollback()
print("\nNow try to load the new images, to check it worked!")

47
examples/mogrify.py
View File

@ -1,47 +0,0 @@
# mogrify.py - test all possible simple type mogrifications
# -*- encoding: latin1 -*-
#
# Copyright (C) 2004-2010 Federico Di Gregorio <fog@debian.org>
#
# 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.
#
# 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..
## put in DSN your DSN string
DSN = 'dbname=test'
## don't modify anything below this line (except for experimenting)
import sys, psycopg2
if len(sys.argv) > 1:
DSN = sys.argv[1]
print("Opening connection using dsn:", DSN)
conn = psycopg2.connect(DSN)
print("Encoding for this connection is", conn.encoding)
curs = conn.cursor()
curs.execute("SELECT %(foo)s AS foo", {'foo':'bar'})
curs.execute("SELECT %(foo)s AS foo", {'foo':None})
curs.execute("SELECT %(foo)s AS foo", {'foo':True})
curs.execute("SELECT %(foo)s AS foo", {'foo':42})
curs.execute("SELECT %(foo)s AS foo", {'foo':u'yatt<EFBFBD>!'})
curs.execute("SELECT %(foo)s AS foo", {'foo':u'bar'})
print(curs.mogrify("SELECT %(foo)s AS foo", {'foo':'bar'}))
print(curs.mogrify("SELECT %(foo)s AS foo", {'foo':None}))
print(curs.mogrify("SELECT %(foo)s AS foo", {'foo':True}))
print(curs.mogrify("SELECT %(foo)s AS foo", {'foo':42}))
print(curs.mogrify("SELECT %(foo)s AS foo", {'foo':u'yatt<EFBFBD>!'}))
print(curs.mogrify("SELECT %(foo)s AS foo", {'foo':u'bar'}))
conn.rollback()

126
examples/myfirstrecipe.py
View File

@ -1,126 +0,0 @@
"""
Using a tuple as a bound variable in "SELECT ... IN (...)" clauses
in PostgreSQL using psycopg2
Some time ago someone asked on the psycopg mailing list how to have a
bound variable expand to the right SQL for an SELECT IN clause:
SELECT * FROM atable WHERE afield IN (value1, value2, value3)
with the values to be used in the IN clause to be passed to the cursor
.execute() method in a tuple as a bound variable, i.e.:
in_values = ("value1", "value2", "value3")
curs.execute("SELECT ... IN %s", (in_values,))
psycopg 1 does support typecasting from Python to PostgreSQL (and back)
only for simple types and this problem has no elegant solution (short or
writing a wrapper class returning the pre-quoted text in an __str__
method.
But psycopg2 offers a simple and elegant solution by partially
implementing the Object Adaptation from PEP 246. psycopg2 moves
the type-casting logic into external adapters and a somehow
broken adapt() function.
While the original adapt() takes 3 arguments, psycopg2's one only takes
1: the bound variable to be adapted. The result is an object supporting
a not-yet well defined protocol that we can call ISQLQuote:
class ISQLQuote:
def getquoted(self):
"Returns a quoted string representing the bound variable."
def getbinary(self):
"Returns a binary quoted string representing the bound variable."
def getbuffer(self):
"Returns the wrapped object itself."
__str__ = getquoted
Then one of the functions (usually .getquoted()) is called by psycopg2 at
the right time to obtain the right, sql-quoted representation for the
corresponding bound variable.
The nice part is that the default, built-in adapters, derived from
psycopg 1 tyecasting code can be overridden by the programmer, simply
replacing them in the psycopg.extensions.adapters dictionary.
Then the solution to the original problem is now obvious: write an
adapter that adapts tuple objects into the right SQL string, by calling
recursively adapt() on each element.
psycopg2 development can be tracked on the psycopg mailing list:
http://lists.initd.org/mailman/listinfo/psycopg
"""
# Copyright (C) 2001-2010 Federico Di Gregorio <fog@debian.org>
#
# 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.
#
# 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
import psycopg2.extensions
from psycopg2.extensions import adapt as psycoadapt
from psycopg2.extensions import register_adapter
class AsIs(object):
"""An adapter that just return the object 'as is'.
psycopg 1.99.9 has some optimizations that make impossible to call
adapt() without adding some basic adapters externally. This limitation
will be lifted in a future release.
"""
def __init__(self, obj):
self.__obj = obj
def getquoted(self):
return self.__obj
class SQL_IN(object):
"""Adapt a tuple to an SQL quotable object."""
def __init__(self, seq):
self._seq = seq
def prepare(self, conn):
pass
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
qobjs = [str(psycoadapt(o).getquoted()) for o in self._seq]
return '(' + ', '.join(qobjs) + ')'
__str__ = getquoted
# add our new adapter class to psycopg list of adapters
register_adapter(tuple, SQL_IN)
register_adapter(float, AsIs)
register_adapter(int, AsIs)
# usually we would call:
#
# conn = psycopg.connect("...")
# curs = conn.cursor()
# curs.execute("SELECT ...", (("this", "is", "the", "tuple"),))
#
# but we have no connection to a database right now, so we just check
# the SQL_IN class by calling psycopg's adapt() directly:
if __name__ == '__main__':
print("Note how the string will be SQL-quoted, but the number will not:")
print(psycoadapt(("this is an 'sql quoted' str\\ing", 1, 2.0)))

45
examples/notify.py
View File

@ -1,45 +0,0 @@
# notify.py - example of getting notifies
#
# Copyright (C) 2001-2010 Federico Di Gregorio <fog@debian.org>
#
# 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.
#
# 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.
## put in DSN your DSN string
DSN = 'dbname=test'
## don't modify anything below this line (except for experimenting)
import sys
import select
import psycopg2
from psycopg2.extensions import ISOLATION_LEVEL_AUTOCOMMIT
if len(sys.argv) > 1:
DSN = sys.argv[1]
print("Opening connection using dsn:", DSN)
conn = psycopg2.connect(DSN)
print("Encoding for this connection is", conn.encoding)
conn.set_isolation_level(ISOLATION_LEVEL_AUTOCOMMIT)
curs = conn.cursor()
curs.execute("listen test")
print("Waiting for 'NOTIFY test'")
while 1:
if select.select([conn],[],[],5)==([],[],[]):
print("Timeout")
else:
conn.poll()
while conn.notifies:
print("Got NOTIFY:", conn.notifies.pop())

54
examples/simple.py
View File

@ -1,54 +0,0 @@
# simple.py - very simple example of plain DBAPI-2.0 usage
#
# currently used as test-me-stress-me script for psycopg 2.0
#
# Copyright (C) 2001-2010 Federico Di Gregorio <fog@debian.org>
#
# 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.
#
# 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.
## put in DSN your DSN string
DSN = 'dbname=test'
## don't modify anything below this line (except for experimenting)
class SimpleQuoter(object):
def sqlquote(x=None):
return "'bar'"
import sys
import psycopg2
if len(sys.argv) > 1:
DSN = sys.argv[1]
print("Opening connection using dsn:", DSN)
conn = psycopg2.connect(DSN)
print("Encoding for this connection is", conn.encoding)
curs = conn.cursor()
curs.execute("SELECT 1 AS foo")
print(curs.fetchone())
curs.execute("SELECT 1 AS foo")
print(curs.fetchmany())
curs.execute("SELECT 1 AS foo")
print(curs.fetchall())
conn.rollback()
sys.exit(0)
curs.execute("SELECT 1 AS foo", async=1)
curs.execute("SELECT %(foo)s AS foo", {'foo':'bar'})
curs.execute("SELECT %(foo)s AS foo", {'foo':None})
curs.execute("SELECT %(foo)f AS foo", {'foo':42})
curs.execute("SELECT %(foo)s AS foo", {'foo':SimpleQuoter()})

BIN
examples/somehackers.jpg
View File

Binary file not shown.
Side by Side

Before

Width:  |  Height:  |  Size: 22 KiB

161
examples/threads.py
View File

@ -1,161 +0,0 @@
# threads.py -- example of multiple threads using psycopg
# -*- encoding: latin1 -*-
#
# Copyright (C) 2001-2010 Federico Di Gregorio <fog@debian.org>
#
# 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.
#
# 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.
## put in DSN your DSN string
DSN = 'dbname=test'
## some others parameters
INSERT_THREADS = ('A', 'B', 'C')
SELECT_THREADS = ('1', '2')
ROWS = 1000
COMMIT_STEP = 20
SELECT_SIZE = 10000
SELECT_STEP = 500
SELECT_DIV = 250
# the available modes are:
# 0 - one connection for all inserts and one for all select threads
# 1 - connections generated using the connection pool
MODE = 1
## don't modify anything below this line (except for experimenting)
import sys, psycopg2, threading
from psycopg2.pool import ThreadedConnectionPool
from psycopg2.extensions import ISOLATION_LEVEL_AUTOCOMMIT
if len(sys.argv) > 1:
DSN = sys.argv[1]
if len(sys.argv) > 2:
MODE = int(sys.argv[2])
print("Opening connection using dsn:", DSN)
conn = psycopg2.connect(DSN)
curs = conn.cursor()
try:
curs.execute("""CREATE TABLE test_threads (
name text, value1 int4, value2 float)""")
except:
conn.rollback()
curs.execute("DROP TABLE test_threads")
curs.execute("""CREATE TABLE test_threads (
name text, value1 int4, value2 float)""")
conn.commit()
## this function inserts a big number of rows and creates and destroys
## a large number of cursors
def insert_func(conn_or_pool, rows):
name = threading.currentThread().getName()
if MODE == 0:
conn = conn_or_pool
else:
conn = conn_or_pool.getconn()
for i in range(rows):
if divmod(i, COMMIT_STEP)[1] == 0:
conn.commit()
if MODE == 1:
conn_or_pool.putconn(conn)
s = name + ": COMMIT STEP " + str(i)
print(s)
if MODE == 1:
conn = conn_or_pool.getconn()
c = conn.cursor()
try:
c.execute("INSERT INTO test_threads VALUES (%s, %s, %s)",
(str(i), i, float(i)))
except psycopg2.ProgrammingError as err:
print(name, ": an error occurred; skipping this insert")
print(err)
conn.commit()
## a nice select function that prints the current number of rows in the
## database (and transfer them, putting some pressure on the network)
def select_func(conn_or_pool, z):
name = threading.currentThread().getName()
if MODE == 0:
conn = conn_or_pool
conn.set_isolation_level(ISOLATION_LEVEL_AUTOCOMMIT)
for i in range(SELECT_SIZE):
if divmod(i, SELECT_STEP)[1] == 0:
try:
if MODE == 1:
conn = conn_or_pool.getconn()
conn.set_isolation_level(ISOLATION_LEVEL_AUTOCOMMIT)
c = conn.cursor()
c.execute("SELECT * FROM test_threads WHERE value2 < %s",
(int(i/z),))
l = c.fetchall()
if MODE == 1:
conn_or_pool.putconn(conn)
s = name + ": number of rows fetched: " + str(len(l))
print(s)
except psycopg2.ProgrammingError as err:
print(name, ": an error occurred; skipping this select")
print(err)
## create the connection pool or the connections
if MODE == 0:
conn_insert = psycopg2.connect(DSN)
conn_select = psycopg2.connect(DSN)
else:
m = len(INSERT_THREADS) + len(SELECT_THREADS)
n = m/2
conn_insert = conn_select = ThreadedConnectionPool(n, m, DSN)
## create the threads
threads = []
print("Creating INSERT threads:")
for name in INSERT_THREADS:
t = threading.Thread(None, insert_func, 'Thread-'+name,
(conn_insert, ROWS))
t.setDaemon(0)
threads.append(t)
print("Creating SELECT threads:")
for name in SELECT_THREADS:
t = threading.Thread(None, select_func, 'Thread-'+name,
(conn_select, SELECT_DIV))
t.setDaemon(0)
threads.append(t)
## really start the threads now
for t in threads:
t.start()
# and wait for them to finish
for t in threads:
t.join()
print(t.getName(), "exited OK")
conn.commit()
curs.execute("SELECT count(name) FROM test_threads")
print("Inserted", curs.fetchone()[0], "rows.")
curs.execute("DROP TABLE test_threads")
conn.commit()

65
examples/typecast.py
View File

@ -1,65 +0,0 @@
# typecast.py - example of per-cursor and per-connection typecasters.
#
# Copyright (C) 2001-2010 Federico Di Gregorio <fog@debian.org>
#
# 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.
#
# 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.
## put in DSN your DSN string
DSN = 'dbname=test'
## don't modify anything below this line (except for experimenting)
class SimpleQuoter(object):
def sqlquote(x=None):
return "'bar'"
import sys
import psycopg2
import psycopg2.extensions
if len(sys.argv) > 1:
DSN = sys.argv[1]
print("Opening connection using dsn:", DSN)
conn = psycopg2.connect(DSN)
print("Encoding for this connection is", conn.encoding)
curs = conn.cursor()
curs.execute("SELECT 'text'::text AS foo")
textoid = curs.description[0][1]
print("Oid for the text datatype is", textoid)
def castA(s, curs):
if s is not None: return "(A) " + s
TYPEA = psycopg2.extensions.new_type((textoid,), "TYPEA", castA)
def castB(s, curs):
if s is not None: return "(B) " + s
TYPEB = psycopg2.extensions.new_type((textoid,), "TYPEB", castB)
curs = conn.cursor()
curs.execute("SELECT 'some text.'::text AS foo")
print("Some text from plain connection:", curs.fetchone()[0])
psycopg2.extensions.register_type(TYPEA, conn)
curs = conn.cursor()
curs.execute("SELECT 'some text.'::text AS foo")
print("Some text from connection with typecaster:", curs.fetchone()[0])
curs = conn.cursor()
psycopg2.extensions.register_type(TYPEB, curs)
curs.execute("SELECT 'some text.'::text AS foo")
print("Some text from cursor with typecaster:", curs.fetchone()[0])
curs = conn.cursor()
curs.execute("SELECT 'some text.'::text AS foo")
print("Some text from connection with typecaster again:", curs.fetchone()[0])

69
examples/tz.py
View File

@ -1,69 +0,0 @@
# tz.py - example of datetime objects with time zones
# -*- encoding: utf8 -*-
#
# Copyright (C) 2004-2010 Federico Di Gregorio <fog@debian.org>
#
# 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.
#
# 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.
## put in DSN your DSN string
DSN = 'dbname=test'
## don't modify anything below this line (except for experimenting)
import sys
import psycopg2
import datetime
from psycopg2.tz import ZERO, LOCAL, FixedOffsetTimezone
if len(sys.argv) > 1:
DSN = sys.argv[1]
print("Opening connection using dsn:", DSN)
conn = psycopg2.connect(DSN)
curs = conn.cursor()
try:
curs.execute("CREATE TABLE test_tz (t timestamp with time zone)")
except:
conn.rollback()
curs.execute("DROP TABLE test_tz")
curs.execute("CREATE TABLE test_tz (t timestamp with time zone)")
conn.commit()
d = datetime.datetime(1971, 10, 19, 22, 30, 0, tzinfo=LOCAL)
curs.execute("INSERT INTO test_tz VALUES (%s)", (d,))
print("Inserted timestamp with timezone:", d)
print("Time zone:", d.tzinfo.tzname(d), "offset:", d.tzinfo.utcoffset(d))
tz = FixedOffsetTimezone(-5*60, "EST")
d = datetime.datetime(1971, 10, 19, 22, 30, 0, tzinfo=tz)
curs.execute("INSERT INTO test_tz VALUES (%s)", (d,))
print("Inserted timestamp with timezone:", d)
print("Time zone:", d.tzinfo.tzname(d), "offset:", d.tzinfo.utcoffset(d))
curs.execute("SELECT * FROM test_tz")
d = curs.fetchone()[0]
curs.execute("INSERT INTO test_tz VALUES (%s)", (d,))
print("Inserted SELECTed timestamp:", d)
print("Time zone:", d.tzinfo.tzname(d), "offset:", d.tzinfo.utcoffset(d))
curs.execute("SELECT * FROM test_tz")
for d in curs:
u = d[0].utcoffset() or ZERO
print("UTC time: ", d[0] - u)
print("Local time:", d[0])
print("Time zone:", d[0].tzinfo.tzname(d[0]), d[0].tzinfo.utcoffset(d[0]))
curs.execute("DROP TABLE test_tz")
conn.commit()

126
examples/usercast.py
View File

@ -1,126 +0,0 @@
# usercast.py -- example of user defined typecasters
# -*- encoding: latin-1 -*-
#
# Copyright (C) 2001-2010 Federico Di Gregorio <fog@debian.org>
#
# 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.
#
# 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.
## put in DSN your DSN string
DSN = 'dbname=test'
## don't modify anything below this line (except for experimenting)
import sys
import psycopg2
import psycopg2.extensions
import whrandom
# importing psycopg.extras will give us a nice tuple adapter: this is wrong
# because the adapter is meant to be used in SQL IN clauses while we use
# tuples to represent points but it works and the example is about Rect, not
# "Point"
import psycopg2.extras
if len(sys.argv) > 1:
DSN = sys.argv[1]
print("Opening connection using dsn:", DSN)
conn = psycopg2.connect(DSN)
print("Initial encoding for this connection is", conn.encoding)
curs = conn.cursor()
try:
curs.execute("CREATE TABLE test_cast (p1 point, p2 point, b box)")
except:
conn.rollback()
curs.execute("DROP TABLE test_cast")
curs.execute("CREATE TABLE test_cast (p1 point, p2 point, b box)")
conn.commit()
# this is the callable object we use as a typecast (the typecast is
# usually a function, but we use a class, just to demonstrate the
# flexibility of the psycopg casting system
class Rect(object):
"""Very simple rectangle.
Note that we use this type as a data holder, as an adapter of itself for
the ISQLQuote protocol used by psycopg's adapt() (see __confrom__ below)
and eventually as a type-caster for the data extracted from the database
(that's why __init__ takes the curs argument.)
"""
def __init__(self, s=None, curs=None):
"""Init the rectangle from the optional string s."""
self.x = self.y = self.width = self.height = 0.0
if s: self.from_string(s)
def __conform__(self, proto):
"""This is a terrible hack, just ignore proto and return self."""
if proto == psycopg2.extensions.ISQLQuote:
return self
def from_points(self, x0, y0, x1, y1):
"""Init the rectangle from points."""
if x0 > x1: (x0, x1) = (x1, x0)
if y0 > y1: (y0, y1) = (y1, y0)
self.x = x0
self.y = y0
self.width = x1 - x0
self.height = y1 - y0
def from_string(self, s):
"""Init the rectangle from a string."""
seq = eval(s)
self.from_points(seq[0][0], seq[0][1], seq[1][0], seq[1][1])
def getquoted(self):
"""Format self as a string usable by the db to represent a box."""
s = "'((%d,%d),(%d,%d))'" % (
self.x, self.y, self.x + self.width, self.y + self.height)
return s
def show(self):
"""Format a description of the box."""
s = "X: %d\tY: %d\tWidth: %d\tHeight: %d" % (
self.x, self.y, self.width, self.height)
return s
# here we select from the empty table, just to grab the description
curs.execute("SELECT b FROM test_cast WHERE 0=1")
boxoid = curs.description[0][1]
print("Oid for the box datatype is", boxoid)
# and build the user cast object
BOX = psycopg2.extensions.new_type((boxoid,), "BOX", Rect)
psycopg2.extensions.register_type(BOX)
# now insert 100 random data (2 points and a box in each row)
for i in range(100):
p1 = (whrandom.randint(0,100), whrandom.randint(0,100))
p2 = (whrandom.randint(0,100), whrandom.randint(0,100))
b = Rect()
b.from_points(whrandom.randint(0,100), whrandom.randint(0,100),
whrandom.randint(0,100), whrandom.randint(0,100))
curs.execute("INSERT INTO test_cast VALUES ('%(p1)s', '%(p2)s', %(box)s)",
{'box':b, 'p1':p1, 'p2':p2})
print("Added 100 boxed to the database")
# select and print all boxes with at least one point inside
curs.execute("SELECT b FROM test_cast WHERE p1 @ b OR p2 @ b")
boxes = curs.fetchall()
print("Found %d boxes with at least a point inside:" % len(boxes))
for box in boxes:
print(" ", box[0].show())
curs.execute("DROP TABLE test_cast")
conn.commit()

BIN
examples/whereareyou.jpg
View File

Binary file not shown.
Side by Side

Before

Width:  |  Height:  |  Size: 34 KiB

46
psycopg2.cproj
View File

@ -48,26 +48,6 @@
<None Include="setup.cfg" />
<None Include="setup.py" />
<None Include="doc\SUCCESS" />
<None Include="examples\binary.py" />
<None Include="examples\copy_from.py" />
<None Include="examples\copy_to.py" />
<None Include="examples\cursor.py" />
<None Include="examples\dialtone.py" />
<None Include="examples\dict.py" />
<None Include="examples\dt.py" />
<None Include="examples\encoding.py" />
<None Include="examples\fetch.py" />
<None Include="examples\lastrowid.py" />
<None Include="examples\mogrify.py" />
<None Include="examples\myfirstrecipe.py" />
<None Include="examples\notify.py" />
<None Include="examples\simple.py" />
<None Include="examples\somehackers.jpg" />
<None Include="examples\threads.py" />
<None Include="examples\typecast.py" />
<None Include="examples\tz.py" />
<None Include="examples\usercast.py" />
<None Include="examples\whereareyou.jpg" />
<None Include="lib\errorcodes.py" />
<None Include="lib\extensions.py" />
<None Include="lib\extras.py" />
@ -131,7 +111,6 @@
<None Include="tests\test_lobject.py" />
<None Include="tests\test_quote.py" />
<None Include="psycopg\lobject.h" />
<None Include="sandbox\test_isready_connection_closed.py" />
<None Include="psycopg\adapter_pfloat.h" />
<None Include="psycopg2da\adapter.py" />
<None Include="psycopg2da\configure.zcml" />
@ -172,33 +151,8 @@
<None Include="tests\test_green.py" />
<None Include="tests\test_notify.py" />
<None Include="scripts\make_errorcodes.py" />
<None Include="sandbox\array.py" />
<None Include="sandbox\async.py" />
<None Include="sandbox\crash.py" />
<None Include="sandbox\domainoid.py" />
<None Include="sandbox\gtk.py" />
<None Include="sandbox\iter.py" />
<None Include="sandbox\leak.test.py" />
<None Include="sandbox\misc_dbapi_test.py" />
<None Include="sandbox\named.py" />
<None Include="sandbox\pbool.py" />
<None Include="sandbox\stress.py" />
<None Include="sandbox\stress2.py" />
<None Include="sandbox\test.py" />
<None Include="sandbox\test814.py" />
<None Include="sandbox\test_copy2.csv" />
<None Include="sandbox\test_copy2.py" />
<None Include="sandbox\test_notices.py" />
<None Include="sandbox\test-psycopg2-datetime-systematic.py" />
<None Include="sandbox\textfloat.py" />
<None Include="sandbox\trigger-commit-fail.py" />
<None Include="sandbox\trigger_double_dealloc.py" />
<None Include="sandbox\typecrash.py" />
<None Include="sandbox\tzhalf.py" />
<None Include="sandbox\valgrind-python.supp" />
<None Include="psycopg\green.h" />
<None Include="doc\src\pool.rst" />
<None Include="sandbox\dec2float.py" />
<None Include="psycopg\notify.h" />
<None Include="psycopg\xid.h" />
<None Include="tests\dbapi20_tpc.py" />

2
tox.ini
View File

@ -8,4 +8,4 @@ whitelist_externals = make
[flake8]
max-line-length = 85
ignore = E128, W503, E741
exclude = build, doc, examples, tests/dbapi20.py
exclude = build, doc, tests/dbapi20.py