/usr/share/pyshared/adodb/adodb_postgres.py is in python-adodb 2.10-1.1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 | ########################################################################
# Vers 2.10 16 July 2008, (c)2004-2008 John Lim (jlim#natsoft.com) All Rights Reserved
# Released under a BSD-style license. See LICENSE.txt.
# Download: http://adodb.sourceforge.net/#pydownload
########################################################################
import adodb
import psycopg
try:
True, False
except NameError:
# Maintain compatibility with Python 2.2
True, False = 1, 0
# Thread Safety= 2 connections
# Param Style = pyformat "%(name)s"
class adodb_postgres(adodb.ADOConnection):
databaseType = 'postgres'
dataProvider = 'postgres'
sysDate = "CURRENT_DATE"
sysTimeStamp = "CURRENT_TIMESTAMP"
metaColSQL = """SELECT a.attname,t.typname,a.attlen,a.atttypmod,a.attnotnull,a.atthasdef,a.attnum
FROM pg_class c, pg_attribute a,pg_type t
WHERE relkind = 'r' AND (c.relname='%s' or c.relname = lower('%s')) and a.attname not like '....%%'
AND a.attnum > 0 AND a.atttypid = t.oid AND a.attrelid = c.oid ORDER BY a.attnum"""
def __init__(self):
pass
def Module(self):
return psycopg
#host=host1 user=user1 password=secret port=4341
def _connect(self,host=None,user=None,password=None,database=None):
if user == None and password == None and database == None:
dsn = host
else:
dsn = 'host='+self.addq(host)
if (user != None): dsn += ' user='+self.addq(user)
if (password != None): dsn += ' password='+self.addq(password)
if (database != None): dsn += ' dbname='+self.addq(database)
self._conn = psycopg.connect(dsn)
self._conn.autocommit(1)
def _newcursor(self,rs):
return cursor_postgres(rs,self)
def SelectLimit(self,sql,limit,offset=-1,params=None):
if (offset >= 0): offset = " OFFSET "+str(offset)
else: offset = ""
return self.Execute(sql+" LIMIT "+str(limit)+offset,params)
def BeginTrans(self):
if self._autocommit:
self._autocommit = False
self._conn.autocommit(0)
def RollbackTrans(self):
self._conn.rollback()
self._autocommit = True
self._conn.autocommit(1)
def CommitTrans(self):
self._conn.commit()
self._autocommit = True
self._conn.autocommit(1)
def _blobencode(self,blob):
blob = str(blob)
#92=backslash, 0=null, 39=single-quote
return blob.replace(chr(92),r'\\134').replace(chr(0),r'\\000').replace(chr(39),r'\\047')
def UpdateBlob(self,table,field,blob,where,blobtype='BLOB'):
if (blobtype == 'BLOB'):
self.Execute("update %s set %s='%s' WHERE %s" % (table,field,self._blobencode(blob),where))
else:
self.Execute("update %s set %s='%s' WHERE %s" % (table,field,self.addq(blob),where))
def MetaColumns(self, table):
#print self.metaColSQL
sql = self.metaColSQL % (table,table)
return self.GetAll(sql)
class cursor_postgres(adodb.ADOCursor):
def __init__(self,rs,conn):
adodb.ADOCursor.__init__(self,rs,conn)
if __name__ == '__main__':
db = adodb_postgres()
db.Connect('localhost','tester','test','test')
adodb.Test(db)
#adodb.Test_Blob(db)
|