ScoDoc-PE/app/scodoc/notesdb.py

579 lines
16 KiB
Python
Raw Normal View History

2020-09-26 16:19:37 +02:00
# -*- mode: python -*-
# -*- coding: utf-8 -*-
2021-07-12 15:13:10 +02:00
import html
2020-09-26 16:19:37 +02:00
import traceback
import psycopg2
import psycopg2.pool
2021-02-02 14:49:49 +01:00
import psycopg2.extras
2021-06-15 15:38:38 +02:00
from flask import g
import app.scodoc.sco_utils as scu
from app.scodoc.notes_log import log
from app.scodoc.sco_exceptions import ScoException, ScoValueError, NoteProcessError
2020-09-26 16:19:37 +02:00
import datetime
2021-07-12 15:13:10 +02:00
quote_html = html.escape
2020-09-26 16:19:37 +02:00
def quote_dict(d):
"html quote all values in dict"
for k in d.keys():
v = d[k]
2021-07-11 18:18:44 +02:00
if isinstance(v, str):
2020-09-26 16:19:37 +02:00
d[k] = quote_html(v, quote=True)
def unquote(s):
"inverse of quote"
# pas d'inverse de cgi.escape
# ne traite que &
# XX voir aussi sco_utils.unescape_html
return s.replace("&", "&")
2021-07-28 09:51:18 +02:00
def open_dept_connection(scodoc_dept=None):
"""Open a connection to the current dept db (g.scodoc_dept)
or to the argument scodoc_dept
"""
return psycopg2.connect(scu.get_db_cnx_string(scodoc_dept))
def close_dept_connection():
"""Commit and close dept db."""
2021-07-19 19:53:01 +02:00
# log("close_dept_connection to " + scu.get_db_cnx_string())
g.db_conn.commit()
g.db_conn.close()
2021-07-28 09:51:18 +02:00
def set_sco_dept(scodoc_dept):
"""Set "context" to given dept
open db connection
"""
g.scodoc_dept = scodoc_dept
if hasattr(g, "db_conn"):
close_dept_connection()
g.db_conn = open_dept_connection()
def GetDBConnexion(autocommit=True): # on n'utilise plus autocommit
return g.db_conn
2020-09-26 16:19:37 +02:00
2021-02-02 14:49:49 +01:00
# Nota: on pourrait maintenant utiliser psycopg2.extras.DictCursor
2020-09-26 16:19:37 +02:00
class ScoDocCursor(psycopg2.extensions.cursor):
"""A database cursor emulating some methods of psycopg v1 cursors"""
2021-02-02 14:49:49 +01:00
def dictfetchall(self):
col_names = [d[0] for d in self.description]
return [dict(zip(col_names, row)) for row in self.fetchall()]
2020-09-26 16:19:37 +02:00
2021-02-02 14:49:49 +01:00
def dictfetchone(self):
col_names = [d[0] for d in self.description]
row = self.fetchone()
2020-09-26 16:19:37 +02:00
if row:
return dict(zip(col_names, row))
else:
return {}
def SimpleQuery(query, args, cursor=None):
2020-09-26 16:19:37 +02:00
if not cursor:
2021-06-15 13:59:56 +02:00
cnx = GetDBConnexion()
2020-09-26 16:19:37 +02:00
cursor = cnx.cursor(cursor_factory=ScoDocCursor)
# log( 'SimpleQuery(%s)' % (query % args) )
cursor.execute(query, args)
return cursor
def SimpleDictFetch(query, args, cursor=None):
cursor = SimpleQuery(query, args, cursor=cursor)
2020-09-26 16:19:37 +02:00
return cursor.dictfetchall()
def DBInsertDict(cnx, table, vals, commit=0, convert_empty_to_nulls=1):
"insert into table values in dict 'vals'"
cursor = cnx.cursor(cursor_factory=ScoDocCursor)
if convert_empty_to_nulls:
for col in vals.keys():
if vals[col] == "":
vals[col] = None
# open('/tmp/vals','a').write( str(vals) + '\n' )
2021-07-09 16:29:45 +02:00
cols = list(vals.keys())
2020-09-26 16:19:37 +02:00
colnames = ",".join(cols)
fmt = ",".join(["%%(%s)s" % col for col in cols])
# print 'insert into %s (%s) values (%s)' % (table,colnames,fmt)
oid = None
try:
if vals:
cursor.execute(
"insert into %s (%s) values (%s)" % (table, colnames, fmt), vals
)
else:
cursor.execute("insert into %s default values" % table)
oid = cursor.lastrowid
except:
log("DBInsertDict: EXCEPTION !")
log("DBInsertDict: table=%s, vals=%s" % (str(table), str(vals)))
log("DBInsertDict: commit (exception)")
cnx.commit() # get rid of this transaction
raise # and re-raise exception
if commit:
# log("DBInsertDict: commit (requested)")
2020-09-26 16:19:37 +02:00
cnx.commit()
return oid
2021-07-12 10:51:45 +02:00
_SQL_REMOVE_BAD_CHARS = str.maketrans("", "", '%*()+=&|[]"`')
2020-09-26 16:19:37 +02:00
def DBSelectArgs(
cnx,
table,
vals,
what=["*"],
sortkey=None,
test="=",
operator="and",
distinct=True,
aux_tables=[],
id_name=None,
limit="",
offset="",
2020-09-26 16:19:37 +02:00
):
"""Select * from table where values match dict vals.
Returns cnx, columns_names, list of tuples
aux_tables = ( tablename, id_name )
"""
cursor = cnx.cursor(cursor_factory=ScoDocCursor)
if sortkey:
orderby = " order by " + sortkey
else:
orderby = ""
if distinct:
distinct = " distinct "
else:
distinct = ""
if limit != "":
limit = " LIMIT %d" % limit
if not offset:
offset = ""
if offset != "":
offset = " OFFSET %d" % offset
2020-09-26 16:19:37 +02:00
operator = " " + operator + " "
# liste des tables (apres "from")
tables = [table] + [x[0] for x in aux_tables]
for i in range(len(tables)):
tables[i] = "%s T%d" % (tables[i], i)
tables = ", ".join(tables)
# condition (apres "where")
cond = ""
i = 1
cl = []
2021-02-05 22:16:30 +01:00
for (_, aux_id) in aux_tables:
2020-09-26 16:19:37 +02:00
cl.append("T0.%s = T%d.%s" % (id_name, i, aux_id))
i = i + 1
cond += " and ".join(cl)
# Traitement des expressions régulières:
# n'autorise pas d'expressions
if test == "~":
for k in vals.keys():
2021-07-12 10:51:45 +02:00
vals[k] = vals[k].translate(_SQL_REMOVE_BAD_CHARS)
2020-09-26 16:19:37 +02:00
if vals:
if aux_tables: # paren
cond += " AND ( "
cond += operator.join(
["T0.%s%s%%(%s)s" % (x, test, x) for x in vals.keys() if vals[x] != None]
)
cnuls = " and ".join(
["T0.%s is NULL" % x for x in vals.keys() if vals[x] is None]
)
if cnuls:
if cond:
cond = cond + " and " + cnuls
else:
cond = cnuls
# close paren
if aux_tables:
cond += ") "
if cond:
cond = " where " + cond
#
req = (
"select "
+ distinct
+ ", ".join(what)
+ " from "
+ tables
+ cond
+ orderby
+ limit
+ offset
)
2020-09-26 16:19:37 +02:00
try:
cursor.execute(req, vals)
except:
log('Exception in DBSelectArgs:\n\treq="%s"\n\tvals="%s"\n' % (req, vals))
log(traceback.format_exc())
raise ScoException()
return cursor.dictfetchall()
def DBUpdateArgs(cnx, table, vals, where=None, commit=False, convert_empty_to_nulls=1):
if not vals or where is None:
return
cursor = cnx.cursor(cursor_factory=ScoDocCursor)
if convert_empty_to_nulls:
for col in vals.keys():
if vals[col] == "":
vals[col] = None
s = ", ".join(["%s=%%(%s)s" % (x, x) for x in vals.keys()])
try:
req = "update " + table + " set " + s + " where " + where
cursor.execute(req, vals)
# log('req=%s\n'%req)
# log('vals=%s\n'%vals)
except:
cnx.commit() # get rid of this transaction
log('Exception in DBUpdateArgs:\n\treq="%s"\n\tvals="%s"\n' % (req, vals))
raise # and re-raise exception
if commit:
cnx.commit()
def DBDelete(cnx, table, colid, val, commit=False):
cursor = cnx.cursor(cursor_factory=ScoDocCursor)
try:
cursor.execute(
"delete from " + table + " where %s=%%(%s)s" % (colid, colid), {colid: val}
)
except:
cnx.commit() # get rid of this transaction
raise # and re-raise exception
if commit:
cnx.commit()
# --------------------------------------------------------------------
2021-07-09 23:31:16 +02:00
class EditableTable(object):
2020-12-26 00:11:55 +01:00
"""--- generic class: SQL table with create/edit/list/delete"""
2020-09-26 16:19:37 +02:00
def __init__(
self,
table_name,
id_name,
dbfields,
sortkey=None,
output_formators={},
input_formators={},
aux_tables=[],
convert_null_outputs_to_empty=True,
allow_set_id=False,
html_quote=True,
fields_creators={}, # { field : [ sql_command_to_create_it ] }
filter_nulls=True, # dont allow to set fields to null
):
self.table_name = table_name
self.id_name = id_name
self.aux_tables = aux_tables
self.dbfields = dbfields
self.sortkey = sortkey
self.output_formators = output_formators
self.input_formators = input_formators
self.convert_null_outputs_to_empty = convert_null_outputs_to_empty
self.allow_set_id = allow_set_id
self.html_quote = html_quote
self.fields_creators = fields_creators
self.filter_nulls = filter_nulls
self.sql_default_values = None
2021-08-01 16:33:09 +02:00
def create(self, cnx, args):
2020-09-26 16:19:37 +02:00
"create object in table"
vals = dictfilter(args, self.dbfields, self.filter_nulls)
2021-07-09 16:29:45 +02:00
if self.id_name in vals and not self.allow_set_id:
2020-09-26 16:19:37 +02:00
del vals[self.id_name]
if self.html_quote:
quote_dict(vals) # quote all HTML markup
# format value
2021-08-01 16:33:09 +02:00
for title in vals:
2021-07-09 16:29:45 +02:00
if title in self.input_formators:
2020-09-26 16:19:37 +02:00
vals[title] = self.input_formators[title](vals[title])
# insert
oid = DBInsertDict(cnx, self.table_name, vals, commit=True)
# get back new object id
cursor = cnx.cursor(cursor_factory=ScoDocCursor)
cursor.execute(
"select %(id_name)s from %(table_name)s where oid=%(oid)s"
% {"id_name": self.id_name, "table_name": self.table_name, "oid": oid}
)
new_id = cursor.fetchone()[0]
return new_id
def delete(self, cnx, oid, commit=True):
"delete tuple"
DBDelete(cnx, self.table_name, self.id_name, oid, commit=commit)
def list(
self,
cnx,
args={},
operator="and",
test="=",
sortkey=None,
disable_formatting=False,
limit="",
offset="",
2020-09-26 16:19:37 +02:00
):
"returns list of dicts"
vals = dictfilter(args, self.dbfields, self.filter_nulls)
if not sortkey:
sortkey = self.sortkey
res = DBSelectArgs(
cnx,
self.table_name,
vals,
sortkey=sortkey,
test=test,
operator=operator,
aux_tables=self.aux_tables,
id_name=self.id_name,
limit=limit,
offset=offset,
2020-09-26 16:19:37 +02:00
)
for r in res:
self.format_output(r, disable_formatting=disable_formatting)
return res
def format_output(self, r, disable_formatting=False):
"Format dict using provided output_formators"
for (k, v) in r.items():
if v is None and self.convert_null_outputs_to_empty:
v = ""
# format value
2021-07-09 16:29:45 +02:00
if not disable_formatting and k in self.output_formators:
2020-09-26 16:19:37 +02:00
try: # XXX debug "isodate"
v = self.output_formators[k](v)
except:
log("*** list: vars=%s" % str(vars()))
log("*** list: r=%s" % str(r))
raise
r[k] = v
def edit(self, cnx, args, html_quote=None):
"""Change fields"""
vals = dictfilter(args, self.dbfields, self.filter_nulls)
html_quote = html_quote or self.html_quote
if html_quote:
quote_dict(vals) # quote HTML
# format value
for title in vals.keys():
2021-07-09 16:29:45 +02:00
if title in self.input_formators:
2020-12-26 00:11:55 +01:00
try:
vals[title] = self.input_formators[title](vals[title])
except:
log("exception while converting %s=%s" % (title, vals[title]))
raise
2020-09-26 16:19:37 +02:00
DBUpdateArgs(
cnx,
self.table_name,
vals,
where="%s=%%(%s)s" % (self.id_name, self.id_name),
commit=True,
)
def dictfilter(d, fields, filter_nulls=True):
2021-08-01 16:33:09 +02:00
"""returns a copy of d with only keys listed in "fields" and non null values"""
2020-09-26 16:19:37 +02:00
r = {}
for f in fields:
2021-07-09 16:29:45 +02:00
if f in d and (d[f] != None or not filter_nulls):
2020-09-26 16:19:37 +02:00
try:
val = d[f].strip()
except:
val = d[f]
r[f] = val
return r
# --------------------------------------------------------------------
# --- Misc Tools
def DateDMYtoISO(dmy, null_is_empty=False):
"convert date string from french format to ISO"
if not dmy:
if null_is_empty:
return ""
else:
return None
2021-07-11 18:18:44 +02:00
if not isinstance(dmy, str):
2020-09-26 16:19:37 +02:00
return dmy.strftime("%Y-%m-%d")
t = dmy.split("/")
if len(t) != 3:
raise ScoValueError('Format de date (j/m/a) invalide: "%s"' % str(dmy))
day, month, year = t
year = int(year)
month = int(month)
day = int(day)
# accept years YYYY or YY, uses 1970 as pivot
if year < 100:
if year > 70:
year += 1900
else:
year += 2000
if month < 1 or month > 12:
raise ScoValueError("mois de la date invalide ! (%s)" % month)
# compute nb of day in month:
mo = month
if mo > 7:
mo = mo + 1
if mo % 2:
MonthNbDays = 31
elif mo == 2:
if year % 4 == 0 and ((year % 100 != 0) or (year % 400 == 0)):
MonthNbDays = 29 # leap
else:
MonthNbDays = 28
else:
MonthNbDays = 30
if day < 1 or day > MonthNbDays:
raise ScoValueError("jour de la date invalide ! (%s)" % day)
return "%04d-%02d-%02d" % (year, month, day)
def DateISOtoDMY(isodate):
if not isodate:
return ""
arg = isodate # debug
# si isodate est une instance de DateTime !
try:
isodate = "%s-%s-%s" % (isodate.year(), isodate.month(), isodate.day())
# log('DateISOtoDMY: converted isodate to iso !')
except:
pass
# drop time from isodate and split
t = str(isodate).split()[0].split("-")
if len(t) != 3:
# XXX recherche bug intermittent assez etrange
log('*** DateISOtoDMY: invalid isodate "%s" (arg="%s")' % (str(isodate), arg))
raise NoteProcessError(
'invalid isodate: "%s" (arg="%s" type=%s)' % (str(isodate), arg, type(arg))
)
year, month, day = t
year = int(year)
month = int(month)
day = int(day)
if month < 1 or month > 12:
raise ValueError("invalid month")
if day < 1 or day > 31:
raise ValueError("invalid day")
return "%02d/%02d/%04d" % (day, month, year)
def TimetoISO8601(t, null_is_empty=False):
"convert time string to ISO 8601 (allow 16:03, 16h03, 16)"
if isinstance(t, datetime.time):
return t.isoformat()
if not t and null_is_empty:
return ""
t = t.strip().upper().replace("H", ":")
if t and t.count(":") == 0 and len(t) < 3:
t = t + ":00"
return t
def TimefromISO8601(t):
"convert time string from ISO 8601 to our display format"
if not t:
return t
# XXX strange bug turnaround...
try:
t = "%s:%s" % (t.hour(), t.minute())
# log('TimefromISO8601: converted isotime to iso !')
except:
pass
fs = str(t).split(":")
return fs[0] + "h" + fs[1] # discard seconds
def TimeDuration(heure_debut, heure_fin):
"""duree (nb entier de minutes) entre deux heures a notre format
ie 12h23
"""
if heure_debut and heure_fin:
h0, m0 = [int(x) for x in heure_debut.split("h")]
h1, m1 = [int(x) for x in heure_fin.split("h")]
d = (h1 - h0) * 60 + (m1 - m0)
return d
else:
return None
def float_null_is_zero(x):
if x is None or x == "":
return 0.0
else:
return float(x)
def int_null_is_zero(x):
if x is None or x == "":
return 0
else:
return int(x)
def int_null_is_null(x):
if x is None or x == "":
return None
else:
return int(x)
def float_null_is_null(x):
if x is None or x == "":
return None
else:
return float(x)
# post filtering
#
def UniqListofDicts(L, key):
"""L is a list of dicts.
Remove from L all items which share the same key/value
"""
# well, code is simpler than documentation:
d = {}
for item in L:
d[item[key]] = item
2021-07-09 16:29:45 +02:00
return list(d.values())
2020-09-26 16:19:37 +02:00
#
def copy_tuples_changing_attribute(
cnx, table, column, old_value, new_value, to_exclude=[]
):
"""Duplicate tuples in DB table, replacing column old_value by new_value
Will raise exception if violation of integerity constraint !
"""
cursor = cnx.cursor(cursor_factory=ScoDocCursor)
cursor.execute(
"select * from %s where %s=%%(old_value)s" % (table, column),
{"old_value": old_value},
)
res = cursor.dictfetchall()
for t in res:
t[column] = new_value
for c in to_exclude:
del t[c]
DBInsertDict(cnx, table, t, convert_empty_to_nulls=False)