PyAPplus64/src/PyAPplus64/applus.py

460 lines
18 KiB
Python

# Copyright (c) 2023 Thomas Tuerk (kontakt@thomas-tuerk.de)
#
# This file is part of PyAPplus64 (see https://www.thomas-tuerk.de/de/pyapplus64).
#
# Use of this source code is governed by an MIT-style
# license that can be found in the LICENSE file or at
# https://opensource.org/licenses/MIT.
from . import applus_db
from . import applus_job
from . import applus_server
from . import applus_sysconf
from . import applus_scripttool
from . import applus_usexml
from . import sql_utils
import yaml
import json
import urllib.parse
from zeep import Client
import pyodbc # type: ignore
from typing import TYPE_CHECKING, Optional, Any, Callable, Dict, Sequence, Set, List
if TYPE_CHECKING:
from _typeshed import FileDescriptorOrPath
class APplusServer:
"""
Verbindung zu einem APplus DB und App Server mit Hilfsfunktionen für den komfortablen Zugriff.
:param db_settings: die Einstellungen für die Verbindung mit der Datenbank
:type db_settings: APplusDBSettings
:param server_settings: die Einstellungen für die Verbindung mit dem APplus App Server
:type server_settings: APplusAppServerSettings
:param web_settings: die Einstellungen für die Verbindung mit dem APplus Web Server
:type web_settings: APplusWebServerSettings
"""
def __init__(self,
db_settings: applus_db.APplusDBSettings,
server_settings: applus_server.APplusServerSettings):
self.db_settings: applus_db.APplusDBSettings = db_settings
"""Die Einstellungen für die Datenbankverbindung"""
self.server_settings : applus_server.APplusServerSettings = server_settings
"""Einstellung für die Verbindung zum APP- und Webserver"""
self._db_conn_pool = list()
"""Eine Liste bestehender DB-Verbindungen"""
self.server_conn: applus_server.APplusServerConnection = applus_server.APplusServerConnection(server_settings)
"""erlaubt den Zugriff auf den AppServer"""
self.sysconf: applus_sysconf.APplusSysConf = applus_sysconf.APplusSysConf(self)
"""erlaubt den Zugriff auf die Sysconfig"""
self.job: applus_job.APplusJob = applus_job.APplusJob(self)
"""erlaubt Arbeiten mit Jobs"""
self.scripttool: applus_scripttool.APplusScriptTool = applus_scripttool.APplusScriptTool(self)
"""erlaubt den einfachen Zugriff auf Funktionen des ScriptTools"""
self._client_table = None
self._client_xml = None
self._client_nummer = None
self._client_adaptdb= None
@property
def client_table(self) -> Client:
if not self._client_table:
self._client_table = self.getAppClient("p2core", "Table")
return self._client_table
@property
def client_xml(self) -> Client:
if not self._client_xml:
self._client_xml = self.getAppClient("p2core", "XML")
return self._client_xml
@property
def client_nummer(self) -> Client:
if not self._client_nummer:
self._client_nummer = self.getAppClient("p2system", "Nummer")
return self._client_nummer
@property
def client_adaptdb(self) -> Client:
if not self._client_adaptdb:
self._client_adaptdb = self.getAppClient("p2dbtools", "AdaptDB")
return self._client_adaptdb
def getDBConnection(self) -> pyodbc.Connection:
"""
Liefert eine pyodbc-Connection zur APplus DB. Diese muss genutzt werden, wenn mehrere Operationen in einer Transaktion
genutzt werden sollen. Ansonsten sind die Hilfsmethoden wie :meth:`APplusServer.dbQuery` zu bevorzugen.
Diese Connection kann in Verbindung mit den Funktionen aus :mod:`PyAPplus64.applus_db` genutzt werden.
Die Verbindung sollte nach Benutzung wieder freigegeben oder geschlossen werden.
"""
if self._db_conn_pool:
return self._db_conn_pool.pop()
else:
conn = self.db_settings.connect()
self._db_conn_pool.append(conn)
return conn
def releaseDBConnection(self, conn : pyodbc.Connection) -> None:
"""Gibt eine DB-Connection zur Wiederverwendung frei"""
self._db_conn_pool.append(conn)
def reconnectDB(self) -> None:
for conn in self._db_conn_pool:
try:
conn.close()
except:
pass
self._db_conn_pool = list()
def completeSQL(self, sql: sql_utils.SqlStatement, raw: bool = False) -> str:
"""
Vervollständigt das SQL-Statement. Es wird z.B. der Mandant hinzugefügt.
:param sql: das SQL Statement
:type sql: sql_utils.SqlStatement
:param raw: soll completeSQL ausgeführt werden? Falls True, wird die Eingabe zurückgeliefert
:type raw: boolean
:return: das vervollständigte SQL-Statement
:rtype: str
"""
if raw:
return str(sql)
else:
return self.client_table.service.getCompleteSQL(sql)
def dbQueryAll(self, sql: sql_utils.SqlStatement, *args: Any, raw: bool = False,
apply: Optional[Callable[[pyodbc.Row], Any]] = None) -> Any:
"""Führt eine SQL Query aus und liefert alle Zeilen zurück. Das SQL wird zunächst
vom Server angepasst, so dass z.B. Mandanteninformation hinzugefügt werden."""
sqlC = self.completeSQL(sql, raw=raw)
conn = self.getDBConnection()
res = applus_db.rawQueryAll(conn, sqlC, *args, apply=apply)
self.releaseDBConnection(conn)
return res
def dbQuerySingleValues(self, sql: sql_utils.SqlStatement, *args: Any, raw: bool = False) -> Sequence[Any]:
"""Führt eine SQL Query aus, die nur eine Spalte zurückliefern soll."""
return self.dbQueryAll(sql, *args, raw=raw, apply=lambda r: r[0])
def dbQuery(self, sql: sql_utils.SqlStatement, f: Callable[[pyodbc.Row], None], *args: Any, raw: bool = False) -> None:
"""Führt eine SQL Query aus und führt für jede Zeile die übergeben Funktion aus.
Das SQL wird zunächst vom Server angepasst, so dass z.B. Mandanteninformation hinzugefügt werden."""
sqlC = self.completeSQL(sql, raw=raw)
conn = self.getDBConnection()
res = applus_db.rawQuery(conn, sqlC, f, *args)
self.releaseDBConnection(conn)
return res
def dbQuerySingleRow(self, sql: sql_utils.SqlStatement, *args: Any, raw: bool = False) -> Optional[pyodbc.Row]:
"""Führt eine SQL Query aus, die maximal eine Zeile zurückliefern soll. Diese Zeile wird geliefert."""
sqlC = self.completeSQL(sql, raw=raw)
conn = self.getDBConnection()
res = applus_db.rawQuerySingleRow(conn, sqlC, *args)
self.releaseDBConnection(conn)
return res
def dbQuerySingleRowDict(self, sql: sql_utils.SqlStatement, *args: Any, raw: bool = False) -> Optional[Dict[str, Any]]:
"""Führt eine SQL Query aus, die maximal eine Zeile zurückliefern soll.
Diese Zeile wird als Dictionary geliefert."""
row = self.dbQuerySingleRow(sql, *args, raw=raw)
if row:
return applus_db.row_to_dict(row)
else:
return None
def dbQuerySingleValue(self, sql: sql_utils.SqlStatement, *args: Any, raw: bool = False) -> Any:
"""Führt eine SQL Query aus, die maximal einen Wert zurückliefern soll.
Dieser Wert oder None wird geliefert."""
sqlC = self.completeSQL(sql, raw=raw)
conn = self.getDBConnection()
res = applus_db.rawQuerySingleValue(conn, sqlC, *args)
self.releaseDBConnection(conn)
return res
def dbExecute(self, sql: sql_utils.SqlStatement, *args: Any, raw: bool = False) -> Any:
"""Führt ein SQL Statement (z.B. update oder insert) aus. Das SQL wird zunächst
vom Server angepasst, so dass z.B. Mandanteninformation hinzugefügt werden."""
sqlC = self.completeSQL(sql, raw=raw)
conn = self.getDBConnection()
res = applus_db.rawExecute(conn, sqlC, *args)
self.releaseDBConnection(conn)
return res
def isDBTableKnown(self, table: str) -> bool:
"""Prüft, ob eine Tabelle im System bekannt ist"""
sql = "select count(*) from SYS.TABLES T where T.NAME=?"
c = self.dbQuerySingleValue(sql, table)
return (c > 0)
def getAppClient(self, package: str, name: str) -> Client:
"""Erzeugt einen zeep - Client für den APP-Server.
Mittels dieses Clients kann eines WSDL Schnittstelle des APP-Servers angesprochen werden.
Wird als *package* "p2core" und als *name* "Table" verwendet und der
resultierende client "client" genannt, dann kann
z.B. mittels "client.service.getCompleteSQL(sql)" vom AppServer ein Vervollständigen
des SQLs angefordert werden.
:param package: das Packet, z.B. "p2core"
:type package: str
:param name: der Name im Packet, z.B. "Table"
:type package: string
:return: den Client
:rtype: Client
"""
return self.server_conn.getAppClient(package, name)
def getWebClient(self, url: str) -> Client:
"""Erzeugt einen zeep - Client für den Web-Server.
Mittels dieses Clients kann die von einer ASMX-Seite zur Verfügung gestellte Schnittstelle angesprochen werden.
Als parameter wird die relative URL der ASMX-Seite erwartet. Die Base-URL automatisch ergänzt.
Ein Beispiel für eine solche relative URL ist "masterdata/artikel.asmx".
ACHTUNG: Als Umgebung wird die Umgebung des sich anmeldenden Nutzers verwendet. Sowohl Nutzer als auch Umgebung können sich von den für App-Clients verwendeten Werten unterscheiden. Wenn möglich, sollte ein App-Client verwendet werden.
:param url: die relative URL der ASMX Seite, z.B. "masterdata/artikel.asmx"
:type package: str
:return: den Client
:rtype: Client
"""
return self.server_conn.getWebClient(url)
def getTableFields(self, table: str, isComputed: Optional[bool] = None) -> Set[str]:
"""
Liefert die Namen aller Felder einer Tabelle.
:param table: Name der Tabelle
:param isComputed: wenn gesetzt, werden nur die Felder geliefert, die berechnet werden oder nicht berechnet werden
:return: Liste aller Feld-Namen
:rtype: {str}
"""
sql = sql_utils.SqlStatementSelect("SYS.TABLES T")
join = sql.addInnerJoin("SYS.COLUMNS C")
join.on.addConditionFieldsEq("T.Object_ID", "C.Object_ID")
if not (isComputed is None):
join.on.addConditionFieldEq("c.is_computed", isComputed)
sql.addFields("C.NAME")
sql.where.addConditionFieldEq("t.name", sql_utils.SqlParam())
return sql_utils.normaliseDBfieldSet(self.dbQueryAll(sql, table, apply=lambda r: r.NAME))
def getUniqueFieldsOfTable(self, table: str) -> Dict[str, List[str]]:
"""
Liefert alle Spalten einer Tabelle, die eindeutig sein müssen.
Diese werden als Dictionary gruppiert nach Index-Namen geliefert.
Jeder Eintrag enthält eine Liste von Feldern, die zusammen eindeutig sein
müssen.
"""
conn = self.getDBConnection()
res = applus_db.getUniqueFieldsOfTable(conn, table)
self.releaseDBConnection(conn)
return res
def useXML(self, xml: str) -> Any:
"""Ruft ``p2core.xml.usexml`` auf. Wird meist durch ein ``UseXMLRow-Objekt`` aufgerufen."""
return self.client_xml.service.useXML(xml)
def mkUseXMLRowInsert(self, table: str) -> applus_usexml.UseXmlRowInsert:
"""
Erzeugt ein Objekt zum Einfügen eines neuen DB-Eintrags.
:param table: DB-Tabelle in die eingefügt werden soll
:type table: str
:return: das XmlRow-Objekt
:rtype: applus_usexml.UseXmlRowInsert
"""
return applus_usexml.UseXmlRowInsert(self, table)
def mkUseXMLRowUpdate(self, table: str, id: int) -> applus_usexml.UseXmlRowUpdate:
return applus_usexml.UseXmlRowUpdate(self, table, id)
def mkUseXMLRowInsertOrUpdate(self, table: str) -> applus_usexml.UseXmlRowInsertOrUpdate:
"""
Erzeugt ein Objekt zum Einfügen oder Updaten eines DB-Eintrags.
:param table: DB-Tabelle in die eingefügt werden soll
:type table: string
:return: das XmlRow-Objekt
:rtype: applus_usexml.UseXmlRowInsertOrUpdate
"""
return applus_usexml.UseXmlRowInsertOrUpdate(self, table)
def mkUseXMLRowDelete(self, table: str, id: int) -> applus_usexml.UseXmlRowDelete:
return applus_usexml.UseXmlRowDelete(self, table, id)
def execUseXMLRowDelete(self, table: str, id: int) -> None:
delRow = self.mkUseXMLRowDelete(table, id)
delRow.delete()
def nextNumber(self, obj: str) -> str:
"""
Erstellt eine neue Nummer für das Objekt und legt diese Nummer zurück.
"""
return self.client_nummer.service.nextNumber(obj)
def updateDatabase(self, file : str) -> str:
"""
Führt eine DBAnpass-xml Datei aus.
:param file: DB-Anpass Datei, die ausgeführt werden soll
:type file: string
:return: Infos zur Ausführung
:rtype: str
"""
jobId = self.job.createSOAPJob("run DBAnpass " + file);
self.client_adaptdb.service.updateDatabase(jobId, "de", file);
res = self.job.getResultURLString(jobId)
if res is None: res = "FEHLER";
if (res == "Folgende Befehle konnten nicht ausgeführt werden:\n\n"):
return ""
else:
return res
def importUdfsAndViews(self, environment : str, views : [str] = [], udfs : [str] = []) -> str:
"""
Importiert bestimmte Views und UDFs
:param environment: die Umgebung, in die Importiert werden soll
:type environment: string
:param views: Views, die importiert werden sollen
:type views: [string]
:param udfs: Views, die importiert werden sollen
:type udfs: [string]
:return: Infos zur Ausführung
:rtype: str
"""
lbl="";
files=[];
for v in views:
files.append({"type" : 1, "name" : v})
for u in udfs:
files.append({"type" : 0, "name" : u})
jobId = self.job.createSOAPJob("importing UDFs and Views");
self.client_adaptdb.service.importUdfsAndViews(jobId, environment, False, json.dumps(files), "de");
res = self.job.getResultURLString(jobId)
if res is None: res = "FEHLER";
return res
def makeWebLink(self, base: str, **kwargs: Any) -> str:
if not self.server_settings.webserver:
raise Exception("keine Webserver-BaseURL gesetzt")
url = str(self.server_settings.webserver) + base
firstArg = True
for arg, argv in kwargs.items():
if not (argv is None):
if firstArg:
firstArg = False
url += "?"
else:
url += "&"
url += arg + "=" + urllib.parse.quote(str(argv))
return url
def makeWebLinkWauftragPos(self, **kwargs: Any) -> str:
return self.makeWebLink("wp/wauftragPosRec.aspx", **kwargs)
def makeWebLinkWauftrag(self, **kwargs: Any) -> str:
return self.makeWebLink("wp/wauftragRec.aspx", **kwargs)
def makeWebLinkBauftrag(self, **kwargs: Any) -> str:
return self.makeWebLink("wp/bauftragRec.aspx", **kwargs)
def makeWebLinkAuftrag(self, **kwargs: Any) -> str:
return self.makeWebLink("sales/auftragRec.aspx", **kwargs)
def makeWebLinkVKRahmen(self, **kwargs: Any) -> str:
return self.makeWebLink("sales/vkrahmenRec.aspx", **kwargs)
def makeWebLinkWarenaugang(self, **kwargs: Any) -> str:
return self.makeWebLink("sales/warenausgangRec.aspx", **kwargs)
def applusFromConfigDict(yamlDict: Dict[str, Any], user: Optional[str] = None, env: Optional[str] = None) -> APplusServer:
"""Läd Einstellungen aus einer Config und erzeugt daraus ein APplus-Objekt"""
if user is None or user == '':
user = yamlDict["appserver"]["user"]
if env is None or env == '':
env = yamlDict["appserver"]["env"]
server_settings = applus_server.APplusServerSettings(
webserver=yamlDict.get("webserver", {}).get("baseurl", None),
appserver=yamlDict["appserver"]["server"],
appserverPort=yamlDict["appserver"]["port"],
user=user, # type: ignore
env=env,
webserverUser=yamlDict.get("webserver", {}).get("user", None),
webserverUserDomain=yamlDict.get("webserver", {}).get("userDomain", None),
webserverPassword=yamlDict.get("webserver", {}).get("password", None))
dbparams = applus_db.APplusDBSettings(
server=yamlDict["dbserver"]["server"],
database=yamlDict["dbserver"]["db"],
user=yamlDict["dbserver"]["user"],
password=yamlDict["dbserver"]["password"])
return APplusServer(dbparams, server_settings)
def applusFromConfigFile(yamlfile: 'FileDescriptorOrPath',
user: Optional[str] = None, env: Optional[str] = None) -> APplusServer:
"""Läd Einstellungen aus einer Config-Datei und erzeugt daraus ein APplus-Objekt"""
yamlDict = {}
with open(yamlfile, "r") as stream:
yamlDict = yaml.safe_load(stream)
return applusFromConfigDict(yamlDict, user=user, env=env)
def applusFromConfig(yamlString: str, user: Optional[str] = None, env: Optional[str] = None) -> APplusServer:
"""Läd Einstellungen aus einer Config-Datei und erzeugt daraus ein APplus-Objekt"""
yamlDict = yaml.safe_load(yamlString)
return applusFromConfigDict(yamlDict, user=user, env=env)
class APplusServerConfigDescription:
"""
Beschreibung einer Configuration bestehend aus Config-Datei, Nutzer und Umgebung.
:param descr: Beschreibung als String, nur für Ausgabe gedacht
:type descr: str
:param yamlfile: die Datei
:type yamlfile: 'FileDescriptorOrPath'
:param user: der Nutzer
:type user: Optional[str]
:param env: die Umgebung
:type env: Optional[str]
"""
def __init__(self,
descr: str,
yamlfile: 'FileDescriptorOrPath',
user:Optional[str] = None,
env:Optional[str] = None):
self.descr = descr
self.yamlfile = yamlfile
self.user = user
self.env = env
def __str__(self) -> str:
return self.descr
def connect(self) -> APplusServer:
return applusFromConfigFile(self.yamlfile, user=self.user, env=self.env)