Browse Source

db and pde modules ok

olivier.massot 8 years ago
parent
commit
39754eba81
2 changed files with 67 additions and 0 deletions
  1. 6 0
      core/db.py
  2. 61 0
      core/pde.py

+ 6 - 0
core/db.py

@@ -53,6 +53,12 @@ class AccessDb(CustomDb):
     def __init__(self, dbpath, **kwargs):
         super(AccessDb, self).__init__(dbq=dbpath, **kwargs)
 
+    def assert_connected(self):
+        for row in self.read("SELECT TOP 1 * FROM MSysObjects;"):
+            if not row:
+                raise AssertionError("Unable to connect to: {}".format(self.connectString))
+            return
+
 class AccessSDb(AccessDb):
     dsn = "DRIVER={Microsoft Access Driver (*.mdb, *.accdb)};FIL={MS Access};"
     default_user = ""

+ 61 - 0
core/pde.py

@@ -0,0 +1,61 @@
+'''
+
+'''
+from path import Path
+
+from core.db import AccessSDb, AccessDb
+
+
+# Web url of the WsPde web service
+WSPDE_URL = "http://localhost:2890/public/WsPDE.asmx"
+
+
+MDW_PATH = r"\\h2o\local\4-transversal\BDD\mda\cg67Parc.mdw"
+UID = "olivier"
+PWD = "massot"
+
+DB_DIRPATH = Path(r"\\h2o\local\4-transversal\BDD\mdb")
+
+FACTURES_DB_PATH = DB_DIRPATH / "Facture_data.mdb"
+CONTROLES_DB_PATH = DB_DIRPATH / "cg67Parc_data.mdb"
+WINCAN_DB_PATH = DB_DIRPATH / r"Wincan\parc_2007\DB\PARC_2007.mdb"
+COMMUN_DB_PATH = DB_DIRPATH / "Commun_Data.mdb"
+ANALYTIQUE_DB_PATH = DB_DIRPATH / "Db_analytique.mdb"
+BO_DB_PATH = DB_DIRPATH / "dbBO.mdb"
+AGRHUM_DB_PATH = DB_DIRPATH / "BDD_ParcRH.mdb"
+
+class ParcDb(AccessSDb):
+    _path = ""
+    def __init__(self, **kwargs):
+        super(AccessSDb, self).__init__(self._path, systemdb=MDW_PATH, uid=UID, pwd=PWD, **kwargs)
+
+class FacturesDb(ParcDb):
+    _path = FACTURES_DB_PATH
+
+class ControlesDb(ParcDb):
+    _path = CONTROLES_DB_PATH
+
+class WincanDb(ParcDb):
+    _path = COMMUN_DB_PATH
+
+class CommunDb(ParcDb):
+    _path = FACTURES_DB_PATH
+
+class AnalytiqueDb(ParcDb):
+    _path = ANALYTIQUE_DB_PATH
+
+class BoDb(ParcDb):
+    _path = BO_DB_PATH
+
+class AgrhumDb(ParcDb):
+    _path = AGRHUM_DB_PATH
+
+
+if __name__ == "__main__":
+    for cls in (FacturesDb, ControlesDb, WincanDb, CommunDb, AnalytiqueDb, BoDb, AgrhumDb):
+        db = cls()
+        db.assert_connected()
+    print("connections ok")
+
+
+