Mercurial > repos > rliterman > csp2

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