Page MenuHomec4science
Files F78255694

dbquery.py.wml
No OneTemporary
Actions

File Metadata

Created
Mon, Aug 19, 10:56

dbquery.py.wml
View Options

## $Id$
## CDSware utility to run SQL queries. The core is taken from
## modpython FAQ and modified to suit our needs. The insert_id() is
## inspired by Erik Forsberg's mod_python slides.
## This file is part of the CERN Document Server Software (CDSware).
## Copyright (C) 2002, 2003, 2004, 2005 CERN.
##
## The CDSware is free software; you can redistribute it and/or
## modify it under the terms of the GNU General Public License as
## published by the Free Software Foundation; either version 2 of the
## License, or (at your option) any later version.
##
## The CDSware is distributed in the hope that it will be useful, but
## WITHOUT ANY WARRANTY; without even the implied warranty of
## MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
## General Public License for more details.
##
## You should have received a copy of the GNU General Public License
## along with CDSware; if not, write to the Free Software Foundation, Inc.,
## 59 Temple Place, Suite 330, Boston, MA 02111-1307, USA.
#include "configbis.wml"
"""CDSware utility to run SQL queries."""
import MySQLdb
import string
## FIXME: these are read here and not in config.py in order to prevent
## them to show up in case some mod_python web script uses ``from
## config import *''. If we waterproof all mod_python web scripts,
## then it may possibly be moved there. But it's safer to read them
## only here.
dbhost = "<DBHOST>"
dbname = "<DBNAME>"
dbuser = "<DBUSER>"
dbpass = "<DBPASS>"
__version__ = "$Id$"
def _db_login(relogin = 0):
"""Login to the database."""
global DB_CONN
if relogin:
DB_CONN = MySQLdb.connect(host=dbhost, db=dbname, user=dbuser, passwd=dbpass)
return DB_CONN
else:
try:
d = DB_CONN
return d
except NameError:
DB_CONN = MySQLdb.connect(host=dbhost, db=dbname, user=dbuser, passwd=dbpass)
return DB_CONN
def run_sql(sql, param=None, n=0, with_desc=0):
"""Run SQL on the server and returns result."""
db = _db_login()
if param:
param = tuple(param)
try:
cur = db.cursor()
rc = cur.execute(sql, param)
except MySQLdb.OperationalError, e: # unexpected disconnect, bad malloc error, etc
# FIXME: now reconnect is always forced, we may perhaps want to ping() first?
db = _db_login(relogin = 1)
cur = db.cursor()
rc = cur.execute(sql, param)
if string.upper(string.split(sql)[0]) in ("SELECT", "SHOW", "DESC", "DESCRIBE"):
if n:
recset = cur.fetchmany(n)
else:
recset = cur.fetchall()
if with_desc:
return recset, cur.description
else:
return recset
else:
if string.upper(string.split(sql)[0]) == "INSERT":
rc = cur.lastrowid
return rc
def blob_to_string(ablob):
"""Return string representation of ABLOB. Useful to treat MySQL
BLOBs in the same way for both recent and old MySQLdb versions.
"""
if type(ablob) is str:
# BLOB is already a string in MySQLdb 0.9.2
return ablob
else:
# BLOB is array.array in MySQLdb 1.0.0 and later
return ablob.tostring()

Event Timeline

c4science · Help