/usr/lib/python2.7/dist-packages/BioSQL/DBUtils.py is in python-biopython-sql 1.66+dfsg-1build1.
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 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 | # Copyright 2002 by Andrew Dalke. All rights reserved.
# Revisions 2007-2010 copyright by Peter Cock. All rights reserved.
# Revisions 2009 copyright by Brad Chapman. All rights reserved.
# Revisions 2013 copyright by Tiago Antao. All rights reserved.
# This code is part of the Biopython distribution and governed by its
# license. Please see the LICENSE file that should have been included
# as part of this package.
#
# Note that BioSQL (including the database schema and scripts) is
# available and licensed separately. Please consult www.biosql.org
import os
_dbutils = {}
class Generic_dbutils(object):
"""Default database utilities."""
def __init__(self):
pass
def tname(self, table):
if table != 'biosequence':
return table
else:
return 'bioentry'
def last_id(self, cursor, table):
# XXX: Unsafe without transactions isolation
table = self.tname(table)
sql = r"select max(%s_id) from %s" % (table, table)
cursor.execute(sql)
rv = cursor.fetchone()
return rv[0]
def execute(self, cursor, sql, args=None):
"""Just execute an sql command.
"""
cursor.execute(sql, args or ())
def autocommit(self, conn, y=1):
# Let's hope it was not really needed
pass
class Sqlite_dbutils(Generic_dbutils):
"""Custom database utilities for SQLite."""
def execute(self, cursor, sql, args=None):
"""Execute SQL command, replacing %s with ? for variable substitution in sqlite3.
"""
cursor.execute(sql.replace("%s", "?"), args or ())
_dbutils["sqlite3"] = Sqlite_dbutils
class Mysql_dbutils(Generic_dbutils):
"""Custom database utilities for MySQL."""
def last_id(self, cursor, table):
if os.name == "java":
return Generic_dbutils.last_id(self, cursor, table)
try:
# This worked on older versions of MySQL
return cursor.insert_id()
except AttributeError:
# See bug 2390
# Google suggests this is the new way,
# same fix also suggested by Eric Gibert:
return cursor.lastrowid
_dbutils["MySQLdb"] = Mysql_dbutils
class _PostgreSQL_dbutils(Generic_dbutils):
"""Base class for any PostgreSQL adaptor."""
def next_id(self, cursor, table):
table = self.tname(table)
sql = r"select nextval('%s_pk_seq')" % table
cursor.execute(sql)
rv = cursor.fetchone()
return rv[0]
def last_id(self, cursor, table):
table = self.tname(table)
sql = r"select currval('%s_pk_seq')" % table
cursor.execute(sql)
rv = cursor.fetchone()
return rv[0]
class Psycopg2_dbutils(_PostgreSQL_dbutils):
"""Custom database utilities for Psycopg2 (PostgreSQL)."""
def autocommit(self, conn, y=True):
if y:
if os.name == "java":
conn.autocommit = 1
else:
conn.set_isolation_level(0)
else:
if os.name == "java":
conn.autocommit = 0
else:
conn.set_isolation_level(1)
_dbutils["psycopg2"] = Psycopg2_dbutils
class Pgdb_dbutils(_PostgreSQL_dbutils):
"""Custom database utilities for Pgdb (aka PyGreSQL, for PostgreSQL)."""
def autocommit(self, conn, y=True):
raise NotImplementedError("pgdb does not support this!")
_dbutils["pgdb"] = Pgdb_dbutils
def get_dbutils(module_name):
try:
return _dbutils[module_name]()
except KeyError:
return Generic_dbutils()
|