jpayne@68: # Copyright 2002 by Andrew Dalke. All rights reserved. jpayne@68: # Revisions 2007-2010 copyright by Peter Cock. All rights reserved. jpayne@68: # Revisions 2009 copyright by Brad Chapman. All rights reserved. jpayne@68: # Revisions 2013 copyright by Tiago Antao. All rights reserved. jpayne@68: # jpayne@68: # This file is part of the Biopython distribution and governed by your jpayne@68: # choice of the "Biopython License Agreement" or the "BSD 3-Clause License". jpayne@68: # Please see the LICENSE file that should have been included as part of this jpayne@68: # package. jpayne@68: # jpayne@68: # Note that BioSQL (including the database schema and scripts) is jpayne@68: # available and licensed separately. Please consult www.biosql.org jpayne@68: """Helper code for Biopython's BioSQL code (for internal use).""" jpayne@68: jpayne@68: import os jpayne@68: from typing import Dict, Type jpayne@68: jpayne@68: jpayne@68: _dbutils: Dict[str, Type["Generic_dbutils"]] = {} jpayne@68: jpayne@68: jpayne@68: class Generic_dbutils: jpayne@68: """Default database utilities.""" jpayne@68: jpayne@68: def __init__(self): jpayne@68: """Create a Generic_dbutils object.""" jpayne@68: jpayne@68: def tname(self, table): jpayne@68: """Return the name of the table.""" jpayne@68: if table != "biosequence": jpayne@68: return table jpayne@68: else: jpayne@68: return "bioentry" jpayne@68: jpayne@68: def last_id(self, cursor, table): jpayne@68: """Return the last used id for a table.""" jpayne@68: # XXX: Unsafe without transactions isolation jpayne@68: table = self.tname(table) jpayne@68: sql = f"select max({table}_id) from {table}" jpayne@68: cursor.execute(sql) jpayne@68: rv = cursor.fetchone() jpayne@68: return rv[0] jpayne@68: jpayne@68: def execute(self, cursor, sql, args=None): jpayne@68: """Just execute an sql command.""" jpayne@68: cursor.execute(sql, args or ()) jpayne@68: jpayne@68: def executemany(self, cursor, sql, seq): jpayne@68: """Execute many sql commands.""" jpayne@68: cursor.executemany(sql, seq) jpayne@68: jpayne@68: def autocommit(self, conn, y=1): jpayne@68: """Set autocommit on the database connection.""" jpayne@68: # Let's hope it was not really needed jpayne@68: jpayne@68: jpayne@68: class Sqlite_dbutils(Generic_dbutils): jpayne@68: """Custom database utilities for SQLite.""" jpayne@68: jpayne@68: def _sub_placeholder(self, sql): jpayne@68: """Format the argument placeholders for sqlite (PRIVATE).""" jpayne@68: return sql.replace("%s", "?") jpayne@68: jpayne@68: def execute(self, cursor, sql, args=None): jpayne@68: """Execute SQL command. jpayne@68: jpayne@68: Replaces %s with ? for variable substitution in sqlite3. jpayne@68: """ jpayne@68: sql = self._sub_placeholder(sql) jpayne@68: cursor.execute(sql, args or ()) jpayne@68: jpayne@68: def executemany(self, cursor, sql, seq): jpayne@68: """Execute many sql statements.""" jpayne@68: sql = self._sub_placeholder(sql) jpayne@68: cursor.executemany(sql, seq) jpayne@68: jpayne@68: jpayne@68: _dbutils["sqlite3"] = Sqlite_dbutils jpayne@68: jpayne@68: jpayne@68: class Mysql_dbutils(Generic_dbutils): jpayne@68: """Custom database utilities for MySQL.""" jpayne@68: jpayne@68: def last_id(self, cursor, table): jpayne@68: """Return the last used id for a table.""" jpayne@68: if os.name == "java": jpayne@68: return Generic_dbutils.last_id(self, cursor, table) jpayne@68: try: jpayne@68: # This worked on older versions of MySQL jpayne@68: return cursor.insert_id() jpayne@68: except AttributeError: jpayne@68: # See bug 2390 jpayne@68: # Google suggests this is the new way, jpayne@68: # same fix also suggested by Eric Gibert: jpayne@68: return cursor.lastrowid jpayne@68: jpayne@68: jpayne@68: _dbutils["MySQLdb"] = Mysql_dbutils jpayne@68: jpayne@68: jpayne@68: class _PostgreSQL_dbutils(Generic_dbutils): jpayne@68: """Base class for any PostgreSQL adaptor.""" jpayne@68: jpayne@68: def next_id(self, cursor, table): jpayne@68: table = self.tname(table) jpayne@68: sql = f"SELECT nextval('{table}_pk_seq')" jpayne@68: cursor.execute(sql) jpayne@68: rv = cursor.fetchone() jpayne@68: return rv[0] jpayne@68: jpayne@68: def last_id(self, cursor, table): jpayne@68: table = self.tname(table) jpayne@68: sql = f"SELECT currval('{table}_pk_seq')" jpayne@68: cursor.execute(sql) jpayne@68: rv = cursor.fetchone() jpayne@68: return rv[0] jpayne@68: jpayne@68: jpayne@68: class Psycopg2_dbutils(_PostgreSQL_dbutils): jpayne@68: """Custom database utilities for Psycopg2 (PostgreSQL).""" jpayne@68: jpayne@68: def autocommit(self, conn, y=True): jpayne@68: """Set autocommit on the database connection.""" jpayne@68: if y: jpayne@68: if os.name == "java": jpayne@68: conn.autocommit = 1 jpayne@68: else: jpayne@68: conn.set_isolation_level(0) jpayne@68: else: jpayne@68: if os.name == "java": jpayne@68: conn.autocommit = 0 jpayne@68: else: jpayne@68: conn.set_isolation_level(1) jpayne@68: jpayne@68: jpayne@68: _dbutils["psycopg2"] = Psycopg2_dbutils jpayne@68: jpayne@68: jpayne@68: class Pgdb_dbutils(_PostgreSQL_dbutils): jpayne@68: """Custom database utilities for Pgdb (aka PyGreSQL, for PostgreSQL).""" jpayne@68: jpayne@68: def autocommit(self, conn, y=True): jpayne@68: """Set autocommit on the database connection. Currently not implemented.""" jpayne@68: raise NotImplementedError("pgdb does not support this!") jpayne@68: jpayne@68: jpayne@68: _dbutils["pgdb"] = Pgdb_dbutils jpayne@68: jpayne@68: jpayne@68: def get_dbutils(module_name): jpayne@68: """Return the correct dbutils object for the database driver.""" jpayne@68: try: jpayne@68: return _dbutils[module_name]() jpayne@68: except KeyError: jpayne@68: return Generic_dbutils()