ScoDoc-Lille/scodoc_manager.py

102 lines
3.1 KiB
Python

# -*- coding: UTF-8 -*
"""
Manage departments, databases.
Each departement `X` has its own database, `SCOX`
and a small configuration file `.../config/depts/X.cfg`
containing the database URI
Old ScoDoc7 installs config files contained `dbname=SCOX`
which translates as
"postgresql://<user>@localhost:5432/<dbname>"
<user> being given by `SCODOC7_SQL_USER` env variable.
"""
import os
import re
import glob
import logging
from flask import g
from config import Config
from app.scodoc.sco_exceptions import ScoConfigurationError
class ScoDeptDescription:
def __init__(self, filename):
"""Read dept description from dept file"""
if os.path.split(filename)[1][-4:] != ".cfg":
raise ScoConfigurationError("Invalid dept config filename: %s" % filename)
self.dept_id = os.path.split(filename)[1][:-4]
if not self.dept_id:
raise ScoConfigurationError("Invalid dept config filename: %s" % filename)
try:
db_uri = open(filename).read().strip()
except:
raise ScoConfigurationError("Department config file missing: %s" % filename)
m = re.match(r"dbname=SCO([a-zA-Z0-9]+$)", db_uri)
if m:
# ScoDoc7 backward compat
dept = m.group(1) # unused in ScoDoc7
db_name = "SCO" + self.dept_id.upper()
db_uri = "postgresql://%(db_user)s@localhost:%(db_port)s/%(db_name)s" % {
"db_user": Config.SCODOC7_SQL_USER,
"db_name": db_name,
"db_port": Config.DEFAULT_SQL_PORT,
}
self.db_uri = db_uri
class ScoDocManager:
def __init__(self):
filenames = glob.glob(Config.SCODOC_VAR_DIR + "/config/depts/*.cfg")
descr_list = [ScoDeptDescription(f) for f in filenames]
self.dept_descriptions = {d.dept_id: d for d in descr_list}
def get_dept_db_uri(self, dept_id):
"DB URI for this dept id"
return self.dept_descriptions[dept_id].db_uri
def get_dept_ids(self):
"get (sorted) dept ids"
return sorted(self.dept_descriptions.keys())
def get_db_uri(self):
"""
Returns DB URI for the "current" departement.
Replaces ScoDoc7 GetDBConnexionString()
"""
return self.get_dept_db_uri(g.scodoc_dept)
sco_mgr = ScoDocManager()
class FakeUsers(object):
"""Temporary for ScoDoc8 devs"""
def __init__(self):
logging.getLogger(__name__).info("Warning: creating a FakeUser instance !")
def user_info(self, user_name="test8", REQUEST=None):
return {
"user_name": user_name,
"nom": user_name,
"prenom": "",
"email": "",
"dept": "",
"nomprenom": user_name,
"prenomnom": user_name,
"prenom_fmt": "",
"nom_fmt": user_name,
"nomcomplet": user_name,
"nomplogin": user_name,
"passwd_temp": 0,
"status": "",
"date_expiration": None,
}
def get_user_list(self, dept=None, with_inactives=False):
return [self.user_info()]