Mercurial > repos > rliterman > csp2

annotate CSP2/CSP2_env/env-d9b9114564458d9d-741b3de822f2aaca6c6caa4325c4afce/lib/python3.8/site-packages/BioSQL/DBUtils.py @ 69:33d812a61356

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
planemo upload commit 2e9511a184a1ca667c7be0c6321a36dc4e3d116d
author jpayne
date Tue, 18 Mar 2025 17:55:14 -0400
parents
children
rev   line source
jpayne@69 1 # Copyright 2002 by Andrew Dalke. All rights reserved.
jpayne@69 2 # Revisions 2007-2010 copyright by Peter Cock. All rights reserved.
jpayne@69 3 # Revisions 2009 copyright by Brad Chapman. All rights reserved.
jpayne@69 4 # Revisions 2013 copyright by Tiago Antao. All rights reserved.
jpayne@69 5 #
jpayne@69 6 # This file is part of the Biopython distribution and governed by your
jpayne@69 7 # choice of the "Biopython License Agreement" or the "BSD 3-Clause License".
jpayne@69 8 # Please see the LICENSE file that should have been included as part of this
jpayne@69 9 # package.
jpayne@69 10 #
jpayne@69 11 # Note that BioSQL (including the database schema and scripts) is
jpayne@69 12 # available and licensed separately. Please consult www.biosql.org
jpayne@69 13 """Helper code for Biopython's BioSQL code (for internal use)."""
jpayne@69 14
jpayne@69 15 import os
jpayne@69 16 from typing import Dict, Type
jpayne@69 17
jpayne@69 18
jpayne@69 19 _dbutils: Dict[str, Type["Generic_dbutils"]] = {}
jpayne@69 20
jpayne@69 21
jpayne@69 22 class Generic_dbutils:
jpayne@69 23 """Default database utilities."""
jpayne@69 24
jpayne@69 25 def __init__(self):
jpayne@69 26 """Create a Generic_dbutils object."""
jpayne@69 27
jpayne@69 28 def tname(self, table):
jpayne@69 29 """Return the name of the table."""
jpayne@69 30 if table != "biosequence":
jpayne@69 31 return table
jpayne@69 32 else:
jpayne@69 33 return "bioentry"
jpayne@69 34
jpayne@69 35 def last_id(self, cursor, table):
jpayne@69 36 """Return the last used id for a table."""
jpayne@69 37 # XXX: Unsafe without transactions isolation
jpayne@69 38 table = self.tname(table)
jpayne@69 39 sql = f"select max({table}_id) from {table}"
jpayne@69 40 cursor.execute(sql)
jpayne@69 41 rv = cursor.fetchone()
jpayne@69 42 return rv[0]
jpayne@69 43
jpayne@69 44 def execute(self, cursor, sql, args=None):
jpayne@69 45 """Just execute an sql command."""
jpayne@69 46 cursor.execute(sql, args or ())
jpayne@69 47
jpayne@69 48 def executemany(self, cursor, sql, seq):
jpayne@69 49 """Execute many sql commands."""
jpayne@69 50 cursor.executemany(sql, seq)
jpayne@69 51
jpayne@69 52 def autocommit(self, conn, y=1):
jpayne@69 53 """Set autocommit on the database connection."""
jpayne@69 54 # Let's hope it was not really needed
jpayne@69 55
jpayne@69 56
jpayne@69 57 class Sqlite_dbutils(Generic_dbutils):
jpayne@69 58 """Custom database utilities for SQLite."""
jpayne@69 59
jpayne@69 60 def _sub_placeholder(self, sql):
jpayne@69 61 """Format the argument placeholders for sqlite (PRIVATE)."""
jpayne@69 62 return sql.replace("%s", "?")
jpayne@69 63
jpayne@69 64 def execute(self, cursor, sql, args=None):
jpayne@69 65 """Execute SQL command.
jpayne@69 66
jpayne@69 67 Replaces %s with ? for variable substitution in sqlite3.
jpayne@69 68 """
jpayne@69 69 sql = self._sub_placeholder(sql)
jpayne@69 70 cursor.execute(sql, args or ())
jpayne@69 71
jpayne@69 72 def executemany(self, cursor, sql, seq):
jpayne@69 73 """Execute many sql statements."""
jpayne@69 74 sql = self._sub_placeholder(sql)
jpayne@69 75 cursor.executemany(sql, seq)
jpayne@69 76
jpayne@69 77
jpayne@69 78 _dbutils["sqlite3"] = Sqlite_dbutils
jpayne@69 79
jpayne@69 80
jpayne@69 81 class Mysql_dbutils(Generic_dbutils):
jpayne@69 82 """Custom database utilities for MySQL."""
jpayne@69 83
jpayne@69 84 def last_id(self, cursor, table):
jpayne@69 85 """Return the last used id for a table."""
jpayne@69 86 if os.name == "java":
jpayne@69 87 return Generic_dbutils.last_id(self, cursor, table)
jpayne@69 88 try:
jpayne@69 89 # This worked on older versions of MySQL
jpayne@69 90 return cursor.insert_id()
jpayne@69 91 except AttributeError:
jpayne@69 92 # See bug 2390
jpayne@69 93 # Google suggests this is the new way,
jpayne@69 94 # same fix also suggested by Eric Gibert:
jpayne@69 95 return cursor.lastrowid
jpayne@69 96
jpayne@69 97
jpayne@69 98 _dbutils["MySQLdb"] = Mysql_dbutils
jpayne@69 99
jpayne@69 100
jpayne@69 101 class _PostgreSQL_dbutils(Generic_dbutils):
jpayne@69 102 """Base class for any PostgreSQL adaptor."""
jpayne@69 103
jpayne@69 104 def next_id(self, cursor, table):
jpayne@69 105 table = self.tname(table)
jpayne@69 106 sql = f"SELECT nextval('{table}_pk_seq')"
jpayne@69 107 cursor.execute(sql)
jpayne@69 108 rv = cursor.fetchone()
jpayne@69 109 return rv[0]
jpayne@69 110
jpayne@69 111 def last_id(self, cursor, table):
jpayne@69 112 table = self.tname(table)
jpayne@69 113 sql = f"SELECT currval('{table}_pk_seq')"
jpayne@69 114 cursor.execute(sql)
jpayne@69 115 rv = cursor.fetchone()
jpayne@69 116 return rv[0]
jpayne@69 117
jpayne@69 118
jpayne@69 119 class Psycopg2_dbutils(_PostgreSQL_dbutils):
jpayne@69 120 """Custom database utilities for Psycopg2 (PostgreSQL)."""
jpayne@69 121
jpayne@69 122 def autocommit(self, conn, y=True):
jpayne@69 123 """Set autocommit on the database connection."""
jpayne@69 124 if y:
jpayne@69 125 if os.name == "java":
jpayne@69 126 conn.autocommit = 1
jpayne@69 127 else:
jpayne@69 128 conn.set_isolation_level(0)
jpayne@69 129 else:
jpayne@69 130 if os.name == "java":
jpayne@69 131 conn.autocommit = 0
jpayne@69 132 else:
jpayne@69 133 conn.set_isolation_level(1)
jpayne@69 134
jpayne@69 135
jpayne@69 136 _dbutils["psycopg2"] = Psycopg2_dbutils
jpayne@69 137
jpayne@69 138
jpayne@69 139 class Pgdb_dbutils(_PostgreSQL_dbutils):
jpayne@69 140 """Custom database utilities for Pgdb (aka PyGreSQL, for PostgreSQL)."""
jpayne@69 141
jpayne@69 142 def autocommit(self, conn, y=True):
jpayne@69 143 """Set autocommit on the database connection. Currently not implemented."""
jpayne@69 144 raise NotImplementedError("pgdb does not support this!")
jpayne@69 145
jpayne@69 146
jpayne@69 147 _dbutils["pgdb"] = Pgdb_dbutils
jpayne@69 148
jpayne@69 149
jpayne@69 150 def get_dbutils(module_name):
jpayne@69 151 """Return the correct dbutils object for the database driver."""
jpayne@69 152 try:
jpayne@69 153 return _dbutils[module_name]()
jpayne@69 154 except KeyError:
jpayne@69 155 return Generic_dbutils()