thtuerk
/
PyAPplus64
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

apply flake8, remove Python 3.10 Syntax 

make sure that the package works with older Python versions:

- replace matches with if-then-else
- Replace "|" with "Union"
- Remove "TypeAbbrev"

Make sure taht flake8 produces few warnings.

Add github action for automatic checks.
This commit is contained in:
Thomas Türk 2023-05-06 21:49:04 +02:00
parent d88469e711
commit 3566c9ba3e
23 changed files with 1191 additions and 1045 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
.flake8 Normal file
View File

@ -0,0 +1,2 @@
[flake8]
max-line-length = 250

43
.github/workflows/python-package.yml vendored Normal file
View File

@ -0,0 +1,43 @@
# This workflow will install Python dependencies, run tests and lint with a variety of Python versions
# For more information see: https://docs.github.com/en/actions/automating-builds-and-tests/building-and-testing-python
name: Python package
on:
push:
branches: [ "main" ]
pull_request:
branches: [ "main" ]
jobs:
build:
runs-on: ubuntu-latest
strategy:
fail-fast: false
matrix:
python-version: ["3.7", "3.8", "3.9", "3.10"]
steps:
- uses: actions/checkout@v3
- name: Set up Python ${{ matrix.python-version }}
uses: actions/setup-python@v3
with:
python-version: ${{ matrix.python-version }}
- name: Install dependencies
run: |
sudo apt update
sudo apt install -y unixodbc
python -m pip install --upgrade pip
python -m pip install flake8 pytest
if [ -f requirements.txt ]; then pip install -r requirements.txt; fi
pip install -e .
- name: Lint with flake8
run: |
# stop the build if there are Python syntax errors or undefined names
flake8 . --count --select=E9,F63,F7,F82 --show-source --statistics
# exit-zero treats all errors as warnings. The GitHub editor is 127 chars wide
flake8 . --count --exit-zero --max-complexity=10 --max-line-length=127 --statistics
- name: Test with pytest
run: |
pytest .

2
docs/source/conf.py
View File

@ -26,7 +26,7 @@ extensions = [
]
templates_path = ['_templates']
exclude_patterns = [] # type: ignore
exclude_patterns = [] # type: ignore
language = 'de'

3
examples/adhoc_report.py
View File

@ -10,7 +10,8 @@ import PyAPplus64
import applus_configs
import pathlib
def main(confFile : pathlib.Path, outfile : str) -> None:
def main(confFile: pathlib.Path, outfile: str) -> None:
server = PyAPplus64.applus.applusFromConfigFile(confFile)
# Einfache SQL-Anfrage

1
examples/applus_configs.py
View File

@ -14,4 +14,3 @@ configdir = basedir.joinpath("config")
serverConfYamlDeploy = configdir.joinpath("applus-server-deploy.yaml")
serverConfYamlTest = configdir.joinpath("applus-server-test.yaml")
serverConfYamlProd = configdir.joinpath("applus-server-prod.yaml")

35
examples/check_dokumente.py
View File

@ -9,26 +9,29 @@
import pathlib
import PyAPplus64
import applus_configs
from typing import Optional
def main(confFile : pathlib.Path, updateDB:bool, docDir:str|None = None) -> None:
server = PyAPplus64.applus.applusFromConfigFile(confFile)
if docDir is None:
docDir = str(server.scripttool.getInstallPathWebServer().joinpath("DocLib"))
def main(confFile: pathlib.Path, updateDB: bool, docDir: Optional[str] = None) -> None:
server = PyAPplus64.applus.applusFromConfigFile(confFile)
sql = PyAPplus64.sql_utils.SqlStatementSelect("ARTIKEL");
sql.addFields("ID", "ARTIKEL", "DOCUMENTS");
sql.where.addConditionFieldStringNotEmpty("DOCUMENTS");
if docDir is None:
docDir = str(server.scripttool.getInstallPathWebServer().joinpath("DocLib"))
for row in server.dbQueryAll(sql):
doc = pathlib.Path(docDir + row.DOCUMENTS);
if not doc.exists():
print("Bild '{}' für Artikel '{}' nicht gefunden".format(doc, row.ARTIKEL))
sql = PyAPplus64.sql_utils.SqlStatementSelect("ARTIKEL")
sql.addFields("ID", "ARTIKEL", "DOCUMENTS")
sql.where.addConditionFieldStringNotEmpty("DOCUMENTS")
for row in server.dbQueryAll(sql):
doc = pathlib.Path(docDir + row.DOCUMENTS)
if not doc.exists():
print("Bild '{}' für Artikel '{}' nicht gefunden".format(doc, row.ARTIKEL))
if updateDB:
upd = server.mkUseXMLRowUpdate("ARTIKEL", row.ID)
upd.addField("DOCUMENTS", None)
upd.update()
if updateDB:
upd = server.mkUseXMLRowUpdate("ARTIKEL", row.ID);
upd.addField("DOCUMENTS", None);
upd.update();
if __name__ == "__main__":
main(applus_configs.serverConfYamlTest, False)
main(applus_configs.serverConfYamlTest, False)

16
examples/copy_artikel.py
View File

@ -24,18 +24,19 @@ import PyAPplus64
import applus_configs
import logging
import yaml
from typing import Optional
def main(confFile:pathlib.Path, artikel:str, artikelNeu:str|None=None) -> None:
def main(confFile: pathlib.Path, artikel: str, artikelNeu: Optional[str] = None) -> None:
# Server verbinden
server = PyAPplus64.applus.applusFromConfigFile(confFile)
# DuplicateBusinessObject für Artikel erstellen
dArt = PyAPplus64.duplicate.loadDBDuplicateArtikel(server, artikel);
dArt = PyAPplus64.duplicate.loadDBDuplicateArtikel(server, artikel)
# DuplicateBusinessObject zur Demonstration in YAML konvertieren und zurück
dArtYaml = yaml.dump(dArt);
print(dArtYaml);
dArtYaml = yaml.dump(dArt)
print(dArtYaml)
dArt2 = yaml.load(dArtYaml, Loader=yaml.UnsafeLoader)
# Neue Artikel-Nummer bestimmen und DuplicateBusinessObject in DB schreiben
@ -44,9 +45,9 @@ def main(confFile:pathlib.Path, artikel:str, artikelNeu:str|None=None) -> None:
artikelNeu = server.nextNumber("Artikel")
if not (dArt is None):
dArt.setFields({"artikel" : artikelNeu})
res = dArt.insert(server);
print(res);
dArt.setFields({"artikel": artikelNeu})
res = dArt.insert(server)
print(res)
if __name__ == "__main__":
@ -56,4 +57,3 @@ if __name__ == "__main__":
# logger.setLevel(logging.ERROR)
main(applus_configs.serverConfYamlTest, "my-artikel", artikelNeu="my-artikel-copy")

143
examples/mengenabweichung.py
View File

@ -11,14 +11,15 @@
import datetime
import PyAPplus64
import applus_configs
import pandas as pd # type: ignore
import pandas as pd # type: ignore
import pathlib
from typing import *
from typing import Tuple, Union, Optional
def ladeAlleWerkstattauftragMengenabweichungen(
server:PyAPplus64.APplusServer,
cond:PyAPplus64.SqlCondition|str|None=None) -> pd.DataFrame:
sql = PyAPplus64.sql_utils.SqlStatementSelect("WAUFTRAG w");
server: PyAPplus64.APplusServer,
cond: Union[PyAPplus64.SqlCondition, str, None] = None) -> pd.DataFrame:
sql = PyAPplus64.sql_utils.SqlStatementSelect("WAUFTRAG w")
sql.addLeftJoin("personal p", "w.UPDUSER = p.PERSONAL")
sql.addFieldsTable("w", "ID", "BAUFTRAG", "POSITION")
@ -30,42 +31,44 @@ def ladeAlleWerkstattauftragMengenabweichungen(
sql.where.addConditionFieldGe("w.STATUS", 5)
sql.where.addCondition("abs(w.MENGE-w.MENGE_IST) > 0.001")
sql.where.addCondition(cond)
sql.order="w.UPDDATE"
dfOrg = PyAPplus64.pandas.pandasReadSql(server, sql);
sql.order = "w.UPDDATE"
dfOrg = PyAPplus64.pandas.pandasReadSql(server, sql)
# Add Links
df = dfOrg.copy();
df = df.drop(columns=["ID"]);
df = dfOrg.copy()
df = df.drop(columns=["ID"])
# df = df[['POSITION', 'BAUFTRAG', 'MENGE']] # reorder / filter columns
df['POSITION'] = PyAPplus64.pandas.mkHyperlinkDataframeColumn(dfOrg,
df['POSITION'] = PyAPplus64.pandas.mkHyperlinkDataframeColumn(
dfOrg,
lambda r: r.POSITION,
lambda r: server.makeWebLinkWauftrag(
bauftrag=r.BAUFTRAG, accessid=r.ID))
df['BAUFTRAG'] = PyAPplus64.pandas.mkHyperlinkDataframeColumn(dfOrg,
df['BAUFTRAG'] = PyAPplus64.pandas.mkHyperlinkDataframeColumn(
dfOrg,
lambda r: r.BAUFTRAG,
lambda r: server.makeWebLinkBauftrag(bauftrag=r.BAUFTRAG))
colNames = {
"BAUFTRAG" : "Betriebsauftrag",
"POSITION" : "Pos",
"MENGENABWEICHUNG" : "Mengenabweichung",
"MENGE" : "Menge",
"MENGE_IST" : "Menge-Ist",
"ARTIKEL" : "Artikel",
"ARTIKELNAME" : "Artikel-Name",
"UPDDATE" : "geändert am",
"UPDNAME" : "geändert von"
"BAUFTRAG": "Betriebsauftrag",
"POSITION": "Pos",
"MENGENABWEICHUNG": "Mengenabweichung",
"MENGE": "Menge",
"MENGE_IST": "Menge-Ist",
"ARTIKEL": "Artikel",
"ARTIKELNAME": "Artikel-Name",
"UPDDATE": "geändert am",
"UPDNAME": "geändert von"
}
df.rename(columns=colNames, inplace=True);
df.rename(columns=colNames, inplace=True)
return df
def ladeAlleWerkstattauftragPosMengenabweichungen(
server : PyAPplus64.APplusServer,
cond:PyAPplus64.SqlCondition|str|None=None) -> pd.DataFrame:
sql = PyAPplus64.sql_utils.SqlStatementSelect("WAUFTRAGPOS w");
server: PyAPplus64.APplusServer,
cond: Union[PyAPplus64.SqlCondition, str, None] = None) -> pd.DataFrame:
sql = PyAPplus64.sql_utils.SqlStatementSelect("WAUFTRAGPOS w")
sql.addLeftJoin("personal p", "w.UPDUSER = p.PERSONAL")
sql.addFieldsTable("w", "ID", "BAUFTRAG", "POSITION", "AG")
@ -76,21 +79,24 @@ def ladeAlleWerkstattauftragPosMengenabweichungen(
sql.where.addConditionFieldEq("w.STATUS", 4)
sql.where.addCondition("abs(w.MENGE-w.MENGE_IST) > 0.001")
sql.where.addCondition(cond)
sql.order="w.UPDDATE"
sql.order = "w.UPDDATE"
dfOrg = PyAPplus64.pandas.pandasReadSql(server, sql);
dfOrg = PyAPplus64.pandas.pandasReadSql(server, sql)
# Add Links
df = dfOrg.copy();
df = df.drop(columns=["ID"]);
df['POSITION'] = PyAPplus64.pandas.mkHyperlinkDataframeColumn(dfOrg,
df = dfOrg.copy()
df = df.drop(columns=["ID"])
df['POSITION'] = PyAPplus64.pandas.mkHyperlinkDataframeColumn(
dfOrg,
lambda r: r.POSITION,
lambda r: server.makeWebLinkWauftrag(
bauftrag=r.BAUFTRAG, accessid=r.ID))
df['BAUFTRAG'] = PyAPplus64.pandas.mkHyperlinkDataframeColumn(dfOrg,
df['BAUFTRAG'] = PyAPplus64.pandas.mkHyperlinkDataframeColumn(
dfOrg,
lambda r: r.BAUFTRAG,
lambda r: server.makeWebLinkBauftrag(bauftrag=r.BAUFTRAG))
df['AG'] = PyAPplus64.pandas.mkHyperlinkDataframeColumn(dfOrg,
df['AG'] = PyAPplus64.pandas.mkHyperlinkDataframeColumn(
dfOrg,
lambda r: r.AG,
lambda r: server.makeWebLinkWauftragPos(
bauftrag=r.BAUFTRAG, position=r.POSITION, accessid=r.ID))
@ -101,21 +107,22 @@ def ladeAlleWerkstattauftragPosMengenabweichungen(
# Rename Columns
colNames = {
"BAUFTRAG" : "Betriebsauftrag",
"POSITION" : "Pos",
"AG" : "AG",
"MENGENABWEICHUNG" : "Mengenabweichung",
"MENGE" : "Menge",
"MENGE_IST" : "Menge-Ist",
"ARTIKEL" : "Artikel",
"UPDDATE" : "geändert am",
"UPDNAME" : "geändert von"
"BAUFTRAG": "Betriebsauftrag",
"POSITION": "Pos",
"AG": "AG",
"MENGENABWEICHUNG": "Mengenabweichung",
"MENGE": "Menge",
"MENGE_IST": "Menge-Ist",
"ARTIKEL": "Artikel",
"UPDDATE": "geändert am",
"UPDNAME": "geändert von"
}
df.rename(columns=colNames, inplace=True);
df.rename(columns=colNames, inplace=True)
return df
def computeInYearMonthCond(field : str, year:int|None=None,
month:int|None=None) -> PyAPplus64.SqlCondition | None:
def computeInYearMonthCond(field: str, year: Optional[int] = None,
month: Optional[int] = None) -> Optional[PyAPplus64.SqlCondition]:
if not (year is None):
if month is None:
return PyAPplus64.sql_utils.SqlConditionDateTimeFieldInYear(field, year)
@ -124,59 +131,67 @@ def computeInYearMonthCond(field : str, year:int|None=None,
else:
return None
def computeFileName(year:int|None=None, month:int|None=None) -> str:
def computeFileName(year: Optional[int] = None, month: Optional[int] = None) -> str:
if year is None:
return 'mengenabweichungen-all.xlsx';
return 'mengenabweichungen-all.xlsx'
else:
if month is None:
return 'mengenabweichungen-{:04d}.xlsx'.format(year);
return 'mengenabweichungen-{:04d}.xlsx'.format(year)
else:
return 'mengenabweichungen-{:04d}-{:02d}.xlsx'.format(year, month);
return 'mengenabweichungen-{:04d}-{:02d}.xlsx'.format(year, month)
def _exportInternal(server: PyAPplus64.APplusServer, fn:str,
cond:Union[PyAPplus64.SqlCondition, str, None]) -> int:
def _exportInternal(server: PyAPplus64.APplusServer, fn: str,
cond: Union[PyAPplus64.SqlCondition, str, None]) -> int:
df1 = ladeAlleWerkstattauftragMengenabweichungen(server, cond)
df2 = ladeAlleWerkstattauftragPosMengenabweichungen(server, cond)
print ("erzeuge " + fn);
print("erzeuge " + fn)
PyAPplus64.pandas.exportToExcel(fn, [(df1, "WAuftrag"), (df2, "WAuftrag-Pos")], addTable=True)
return len(df1.index) + len(df2.index)
def exportVonBis(server: PyAPplus64.APplusServer, fn:str,
von:datetime.datetime|None, bis:datetime.datetime|None) -> int:
cond = PyAPplus64.sql_utils.SqlConditionDateTimeFieldInRange("w.UPDDATE", von, bis)
return _exportInternal(server, fn, cond)
def exportVonBis(server: PyAPplus64.APplusServer, fn: str,
von: Optional[datetime.datetime], bis: Optional[datetime.datetime]) -> int:
cond = PyAPplus64.sql_utils.SqlConditionDateTimeFieldInRange("w.UPDDATE", von, bis)
return _exportInternal(server, fn, cond)
def exportYearMonth(server: PyAPplus64.APplusServer,
year:int|None=None, month:int|None=None) -> int:
cond=computeInYearMonthCond("w.UPDDATE", year=year, month=month)
year: Optional[int] = None, month: Optional[int] = None) -> int:
cond = computeInYearMonthCond("w.UPDDATE", year=year, month=month)
fn = computeFileName(year=year, month=month)
return _exportInternal(server, fn, cond)
def computePreviousMonthYear(cyear : int, cmonth :int) -> Tuple[int, int]:
def computePreviousMonthYear(cyear: int, cmonth: int) -> Tuple[int, int]:
if cmonth == 1:
return (cyear-1, 12)
else:
return (cyear, cmonth-1);
return (cyear, cmonth-1)
def computeNextMonthYear(cyear : int, cmonth :int) -> Tuple[int, int]:
def computeNextMonthYear(cyear: int, cmonth: int) -> Tuple[int, int]:
if cmonth == 12:
return (cyear+1, 1)
else:
return (cyear, cmonth+1);
return (cyear, cmonth+1)
def main(confFile : str|pathlib.Path, user:str|None=None, env:str|None=None) -> None:
def main(confFile: Union[str, pathlib.Path], user: Optional[str] = None, env: Optional[str] = None) -> None:
server = PyAPplus64.applusFromConfigFile(confFile, user=user, env=env)
now = datetime.date.today()
(cmonth, cyear) = (now.month, now.year)
(pyear, pmonth) = computePreviousMonthYear(cyear, cmonth);
(pyear, pmonth) = computePreviousMonthYear(cyear, cmonth)
# Ausgaben
exportYearMonth(server, cyear, cmonth) # Aktueller Monat
exportYearMonth(server, pyear, pmonth) # Vorheriger Monat
exportYearMonth(server, cyear, cmonth) # Aktueller Monat
exportYearMonth(server, pyear, pmonth) # Vorheriger Monat
# export(cyear) # aktuelles Jahr
# export(cyear-1) # letztes Jahr
# export() # alles
if __name__ == "__main__":
main(applus_configs.serverConfYamlTest)

54
examples/mengenabweichung_gui.pyw
View File

@ -6,15 +6,16 @@
# license that can be found in the LICENSE file or at
# https://opensource.org/licenses/MIT.
import PySimpleGUI as sg # type: ignore
import PySimpleGUI as sg # type: ignore
import mengenabweichung
import datetime
import PyAPplus64
import applus_configs
import pathlib
from typing import *
from typing import Tuple, Optional, Union
def parseDate (dateS:str) -> Tuple[datetime.datetime|None, bool]:
def parseDate(dateS: str) -> Tuple[Optional[datetime.datetime], bool]:
if dateS is None or dateS == '':
return (None, True)
else:
@ -24,12 +25,15 @@ def parseDate (dateS:str) -> Tuple[datetime.datetime|None, bool]:
sg.popup_error("Fehler beim Parsen des Datums '{}'".format(dateS))
return (None, False)
def createFile(server:PyAPplus64.APplusServer, fileS:str, vonS:str, bisS:str)->None:
def createFile(server: PyAPplus64.APplusServer, fileS: str, vonS: str, bisS: str) -> None:
(von, vonOK) = parseDate(vonS)
if not vonOK: return
if not vonOK:
return
(bis, bisOK) = parseDate(bisS)
if not bisOK: return
if not bisOK:
return
if (fileS is None) or fileS == '':
sg.popup_error("Es wurde keine Ausgabedatei ausgewählt.")
@ -41,22 +45,23 @@ def createFile(server:PyAPplus64.APplusServer, fileS:str, vonS:str, bisS:str)->N
sg.popup_ok("{} Datensätze erfolgreich in Datei '{}' geschrieben.".format(c, file))
def main(confFile : str|pathlib.Path, user:str|None=None, env:str|None=None) -> None:
def main(confFile: Union[str, pathlib.Path], user: Optional[str] = None, env: Optional[str] = None) -> None:
server = PyAPplus64.applusFromConfigFile(confFile, user=user, env=env)
layout = [
[sg.Text(('Bitte geben Sie an, für welchen Zeitraum die '
'Mengenabweichungen ausgegeben werden sollen:'))],
[sg.Text('Von (einschließlich)', size=(15,1)), sg.InputText(key='Von'),
[sg.Text('Von (einschließlich)', size=(15, 1)), sg.InputText(key='Von'),
sg.CalendarButton("Kalender", close_when_date_chosen=True,
target="Von", format='%d.%m.%Y')],
[sg.Text('Bis (ausschließlich)', size=(15,1)), sg.InputText(key='Bis'),
[sg.Text('Bis (ausschließlich)', size=(15, 1)), sg.InputText(key='Bis'),
sg.CalendarButton("Kalender", close_when_date_chosen=True,
target="Bis", format='%d.%m.%Y')],
[sg.Text('Ausgabedatei', size=(15,1)), sg.InputText(key='File'),
sg.FileSaveAs(button_text="wählen", target="File",
file_types = (('Excel Files', '*.xlsx'),),
default_extension = ".xlsx")],
[sg.Text('Ausgabedatei', size=(15, 1)), sg.InputText(key='File'),
sg.FileSaveAs(button_text="wählen",
target="File",
file_types=(('Excel Files', '*.xlsx'),),
default_extension=".xlsx")],
[sg.Button("Aktueller Monat"), sg.Button("Letzter Monat"),
sg.Button("Aktuelles Jahr"), sg.Button("Letztes Jahr")],
[sg.Button("Speichern"), sg.Button("Beenden")]
@ -66,33 +71,34 @@ def main(confFile : str|pathlib.Path, user:str|None=None, env:str|None=None) ->
window = sg.Window("Mengenabweichung " + systemName, layout)
now = datetime.date.today()
(cmonth, cyear) = (now.month, now.year)
(pyear, pmonth) = mengenabweichung.computePreviousMonthYear(cyear, cmonth);
(nyear, nmonth) = mengenabweichung.computeNextMonthYear(cyear, cmonth);
(pyear, pmonth) = mengenabweichung.computePreviousMonthYear(cyear, cmonth)
(nyear, nmonth) = mengenabweichung.computeNextMonthYear(cyear, cmonth)
while True:
event, values = window.read()
if event == sg.WIN_CLOSED or event == 'Beenden':
break
if event == 'Aktueller Monat':
window['Von'].update(value="01.{:02d}.{:04d}".format(cmonth, cyear));
window['Bis'].update(value="01.{:02d}.{:04d}".format(nmonth, nyear));
window['Von'].update(value="01.{:02d}.{:04d}".format(cmonth, cyear))
window['Bis'].update(value="01.{:02d}.{:04d}".format(nmonth, nyear))
if event == 'Letzter Monat':
window['Von'].update(value="01.{:02d}.{:04d}".format(pmonth, pyear));
window['Bis'].update(value="01.{:02d}.{:04d}".format(cmonth, cyear));
window['Von'].update(value="01.{:02d}.{:04d}".format(pmonth, pyear))
window['Bis'].update(value="01.{:02d}.{:04d}".format(cmonth, cyear))
if event == 'Aktuelles Jahr':
window['Von'].update(value="01.01.{:04d}".format(cyear));
window['Bis'].update(value="01.01.{:04d}".format(cyear+1));
window['Von'].update(value="01.01.{:04d}".format(cyear))
window['Bis'].update(value="01.01.{:04d}".format(cyear+1))
if event == 'Letztes Jahr':
window['Von'].update(value="01.01.{:04d}".format(cyear-1));
window['Bis'].update(value="01.01.{:04d}".format(cyear));
window['Von'].update(value="01.01.{:04d}".format(cyear-1))
window['Bis'].update(value="01.01.{:04d}".format(cyear))
if event == 'Speichern':
try:
createFile(server, values.get('File', None),
values.get('Von', None), values.get('Bis', None))
except Exception as e:
sg.popup_error_with_traceback("Beim Erzeugen der Excel-Datei trat ein Fehler auf:", e);
sg.popup_error_with_traceback("Beim Erzeugen der Excel-Datei trat ein Fehler auf:", e)
window.close()
if __name__ == "__main__":
main(applus_configs.serverConfYamlProd)

55
examples/read_settings.py
View File

@ -13,40 +13,41 @@
import pathlib
import PyAPplus64
import applus_configs
import lxml.etree as ET # type: ignore
from typing import Optional, Union
def main(confFile : str|pathlib.Path, user:str|None=None, env:str|None=None) -> None:
server = PyAPplus64.applusFromConfigFile(confFile, user=user, env=env)
print ("\n\nSysConf Lookups:")
def main(confFile: Union[str, pathlib.Path], user: Optional[str] = None, env: Optional[str] = None) -> None:
server = PyAPplus64.applusFromConfigFile(confFile, user=user, env=env)
print (" Default Auftragsart:", server.sysconf.getString("STAMM", "DEFAULTAUFTRAGSART"))
print (" Auftragsarten:")
arten = server.sysconf.getList("STAMM", "AUFTRAGSART", sep='\n')
if not arten: arten = []
for a in arten:
print (" - " + a)
print("\n\nSysConf Lookups:")
print (" Firmen-Nr. automatisch vergeben:", server.sysconf.getBoolean("STAMM", "FIRMAAUTOMATIK"))
print (" Anzahl Artikelstellen:", server.sysconf.getInt("STAMM", "ARTKLASSIFNRLAENGE"))
print(" Default Auftragsart:", server.sysconf.getString("STAMM", "DEFAULTAUFTRAGSART"))
print(" Auftragsarten:")
arten = server.sysconf.getList("STAMM", "AUFTRAGSART", sep='\n')
if not arten:
arten = []
for a in arten:
print(" - " + a)
print ("\n\nScriptTool:")
print(" Firmen-Nr. automatisch vergeben:", server.sysconf.getBoolean("STAMM", "FIRMAAUTOMATIK"))
print(" Anzahl Artikelstellen:", server.sysconf.getInt("STAMM", "ARTKLASSIFNRLAENGE"))
print (" CurrentDate:", server.scripttool.getCurrentDate())
print (" CurrentTime:", server.scripttool.getCurrentTime())
print (" CurrentDateTime:", server.scripttool.getCurrentDateTime())
print (" LoginName:", server.scripttool.getLoginName())
print (" UserName:", server.scripttool.getUserName())
print (" UserFullName:", server.scripttool.getUserFullName())
print (" SystemName:", server.scripttool.getSystemName())
print (" Mandant:", server.scripttool.getMandant())
print (" MandantName:", server.scripttool.getMandantName())
print (" InstallPath:", server.scripttool.getInstallPath())
print (" InstallPathAppServer:", server.scripttool.getInstallPathAppServer())
print (" InstallPathWebServer:", server.scripttool.getInstallPathWebServer())
print (" ServerInfo - Version:", server.scripttool.getServerInfo().find("version").text)
print("\n\nScriptTool:")
print(" CurrentDate:", server.scripttool.getCurrentDate())
print(" CurrentTime:", server.scripttool.getCurrentTime())
print(" CurrentDateTime:", server.scripttool.getCurrentDateTime())
print(" LoginName:", server.scripttool.getLoginName())
print(" UserName:", server.scripttool.getUserName())
print(" UserFullName:", server.scripttool.getUserFullName())
print(" SystemName:", server.scripttool.getSystemName())
print(" Mandant:", server.scripttool.getMandant())
print(" MandantName:", server.scripttool.getMandantName())
print(" InstallPath:", server.scripttool.getInstallPath())
print(" InstallPathAppServer:", server.scripttool.getInstallPathAppServer())
print(" InstallPathWebServer:", server.scripttool.getInstallPathWebServer())
print(" ServerInfo - Version:", server.scripttool.getServerInfo().find("version").text)
if __name__ == "__main__":
main(applus_configs.serverConfYamlTest)
main(applus_configs.serverConfYamlTest)

142
src/PyAPplus64/applus.py
View File

@ -6,8 +6,6 @@
# license that can be found in the LICENSE file or at
# https://opensource.org/licenses/MIT.
#-*- coding: utf-8 -*-
from . import applus_db
from . import applus_server
from . import applus_sysconf
@ -17,14 +15,13 @@ from . import sql_utils
import yaml
import urllib.parse
from zeep import Client
import pyodbc # type: ignore
from typing import *
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.
@ -36,12 +33,15 @@ class APplusServer:
: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.APplusAppServerSettings, web_settings : applus_server.APplusWebServerSettings):
def __init__(self,
db_settings: applus_db.APplusDBSettings,
server_settings: applus_server.APplusAppServerSettings,
web_settings: applus_server.APplusWebServerSettings):
self.db_settings : applus_db.APplusDBSettings = db_settings
self.db_settings: applus_db.APplusDBSettings = db_settings
"""Die Einstellungen für die Datenbankverbindung"""
self.web_settings : applus_server.APplusWebServerSettings = web_settings
self.web_settings: applus_server.APplusWebServerSettings = web_settings
"""Die Einstellungen für die Datenbankverbindung"""
self.db_conn = db_settings.connect()
@ -51,18 +51,17 @@ class APplusServer:
Diese Connection kann in Verbindung mit den Funktionen aus :mod:`PyAPplus64.applus_db` genutzt werden.
"""
self.server_conn : applus_server.APplusServerConnection = applus_server.APplusServerConnection(server_settings);
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);
self.sysconf: applus_sysconf.APplusSysConf = applus_sysconf.APplusSysConf(self)
"""erlaubt den Zugriff auf die Sysconfig"""
self.scripttool : applus_scripttool.APplusScriptTool = applus_scripttool.APplusScriptTool(self);
self.scripttool: applus_scripttool.APplusScriptTool = applus_scripttool.APplusScriptTool(self)
"""erlaubt den einfachen Zugriff auf Funktionen des ScriptTools"""
self.client_table = self.server_conn.getClient("p2core","Table");
self.client_xml = self.server_conn.getClient("p2core","XML");
self.client_table = self.server_conn.getClient("p2core", "Table")
self.client_xml = self.server_conn.getClient("p2core", "XML")
self.client_nummer = self.server_conn.getClient("p2system", "Nummer")
def reconnectDB(self) -> None:
@ -72,7 +71,7 @@ class APplusServer:
pass
self.db_conn = self.db_settings.connect()
def completeSQL(self, sql : sql_utils.SqlStatement, raw:bool=False) -> str:
def completeSQL(self, sql: sql_utils.SqlStatement, raw: bool = False) -> str:
"""
Vervollständigt das SQL-Statement. Es wird z.B. der Mandant hinzugefügt.
@ -86,52 +85,52 @@ class APplusServer:
if raw:
return str(sql)
else:
return self.client_table.service.getCompleteSQL(sql);
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:
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);
sqlC = self.completeSQL(sql, raw=raw)
return applus_db.rawQueryAll(self.db_conn, sqlC, *args, apply=apply)
def dbQuerySingleValues(self, sql : sql_utils.SqlStatement, *args:Any, raw:bool=False) -> Sequence[Any]:
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:
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);
sqlC = self.completeSQL(sql, raw=raw)
applus_db.rawQuery(self.db_conn, sqlC, f, *args)
def dbQuerySingleRow(self, sql:sql_utils.SqlStatement, *args:Any, raw:bool=False) -> Optional[pyodbc.Row]:
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);
sqlC = self.completeSQL(sql, raw=raw)
return applus_db.rawQuerySingleRow(self.db_conn, sqlC, *args)
def dbQuerySingleRowDict(self, sql:sql_utils.SqlStatement, *args:Any, raw:bool=False) -> Optional[Dict[str, Any]]:
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);
row = self.dbQuerySingleRow(sql, *args, raw=raw)
if row:
return applus_db.row_to_dict(row);
return applus_db.row_to_dict(row)
else:
return None
def dbQuerySingleValue(self, sql:sql_utils.SqlStatement, *args:Any, raw:bool=False) -> Any:
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);
sqlC = self.completeSQL(sql, raw=raw)
return applus_db.rawQuerySingleValue(self.db_conn, sqlC, *args)
def isDBTableKnown(self, table : str) -> bool:
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);
c = self.dbQuerySingleValue(sql, table)
return (c > 0)
def getClient(self, package : str, name : str) -> Client:
def getClient(self, package: str, name: str) -> Client:
"""Erzeugt einen zeep - Client.
Mittels dieses Clients kann die WSDL Schnittstelle angesprochen werden.
Wird als *package* "p2core" und als *name* "Table" verwendet und der
@ -146,9 +145,9 @@ class APplusServer:
:return: den Client
:rtype: Client
"""
return self.server_conn.getClient(package, name);
return self.server_conn.getClient(package, name)
def getTableFields(self, table:str, isComputed:Optional[bool]=None) -> Set[str]:
def getTableFields(self, table: str, isComputed: Optional[bool] = None) -> Set[str]:
"""
Liefert die Namen aller Felder einer Tabelle.
@ -158,16 +157,16 @@ class APplusServer:
:rtype: {str}
"""
sql = sql_utils.SqlStatementSelect("SYS.TABLES T")
join = sql.addInnerJoin("SYS.COLUMNS C");
join = sql.addInnerJoin("SYS.COLUMNS C")
join.on.addConditionFieldsEq("T.Object_ID", "C.Object_ID")
if not (isComputed == None):
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));
return sql_utils.normaliseDBfieldSet(self.dbQueryAll(sql, table, apply=lambda r: r.NAME))
def getUniqueFieldsOfTable(self, table : str) -> Dict[str, List[str]]:
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.
@ -176,13 +175,11 @@ class APplusServer:
"""
return applus_db.getUniqueFieldsOfTable(self.db_conn, table)
def useXML(self, xml : str) -> Any:
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);
return self.client_xml.service.useXML(xml)
def mkUseXMLRowInsert(self, table : str) -> applus_usexml.UseXmlRowInsert:
def mkUseXMLRowInsert(self, table: str) -> applus_usexml.UseXmlRowInsert:
"""
Erzeugt ein Objekt zum Einfügen eines neuen DB-Eintrags.
@ -194,10 +191,10 @@ class APplusServer:
return applus_usexml.UseXmlRowInsert(self, table)
def mkUseXMLRowUpdate(self, table : str, id : int) -> applus_usexml.UseXmlRowUpdate:
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:
def mkUseXMLRowInsertOrUpdate(self, table: str) -> applus_usexml.UseXmlRowInsertOrUpdate:
"""
Erzeugt ein Objekt zum Einfügen oder Updaten eines DB-Eintrags.
@ -209,57 +206,55 @@ class APplusServer:
return applus_usexml.UseXmlRowInsertOrUpdate(self, table)
def mkUseXMLRowDelete(self, table:str, id:int) -> applus_usexml.UseXmlRowDelete :
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:
def execUseXMLRowDelete(self, table: str, id: int) -> None:
delRow = self.mkUseXMLRowDelete(table, id)
delRow.delete();
delRow.delete()
def nextNumber(self, obj : str) -> str:
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 makeWebLink(self, base : str, **kwargs : Any) -> str :
def makeWebLink(self, base: str, **kwargs: Any) -> str:
if not self.web_settings.baseurl:
raise Exception("keine Webserver-BaseURL gesetzt");
raise Exception("keine Webserver-BaseURL gesetzt")
url = str(self.web_settings.baseurl) + base;
url = str(self.web_settings.baseurl) + base
firstArg = True
for arg, argv in kwargs.items():
if not (argv == None):
if not (argv is None):
if firstArg:
firstArg = False;
firstArg = False
url += "?"
else:
url += "&"
url += arg + "=" + urllib.parse.quote(str(argv))
return url;
return url
def makeWebLinkWauftragPos(self, **kwargs : Any) -> str:
return self.makeWebLink("wp/wauftragPosRec.aspx", **kwargs);
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 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 makeWebLinkBauftrag(self, **kwargs: Any) -> str:
return self.makeWebLink("wp/bauftragRec.aspx", **kwargs)
def applusFromConfigDict(yamlDict:Dict[str, Any], user:Optional[str]=None, env:Optional[str]=None) -> APplusServer:
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=='':
if user is None or user == '':
user = yamlDict["appserver"]["user"]
if env is None or env=='':
if env is None or env == '':
env = yamlDict["appserver"]["env"]
app_server = applus_server.APplusAppServerSettings(
appserver=yamlDict["appserver"]["server"],
appserverPort=yamlDict["appserver"]["port"],
user=user, # type: ignore
user=user, # type: ignore
env=env)
web_server = applus_server.APplusWebServerSettings(
baseurl=yamlDict.get("webserver", {}).get("baseurl", None)
@ -268,11 +263,12 @@ def applusFromConfigDict(yamlDict:Dict[str, Any], user:Optional[str]=None, env:O
server=yamlDict["dbserver"]["server"],
database=yamlDict["dbserver"]["db"],
user=yamlDict["dbserver"]["user"],
password=yamlDict["dbserver"]["password"]);
return APplusServer(dbparams, app_server, web_server);
password=yamlDict["dbserver"]["password"])
return APplusServer(dbparams, app_server, web_server)
def applusFromConfigFile(yamlfile : 'FileDescriptorOrPath',
user:Optional[str]=None, env:Optional[str]=None) -> APplusServer:
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:
@ -280,8 +276,8 @@ def applusFromConfigFile(yamlfile : 'FileDescriptorOrPath',
return applusFromConfigDict(yamlDict, user=user, env=env)
def applusFromConfig(yamlString : str, user:Optional[str]=None, env:Optional[str]=None) -> APplusServer:
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)

67
src/PyAPplus64/applus_db.py
View File

@ -6,29 +6,27 @@
# license that can be found in the LICENSE file or at
# https://opensource.org/licenses/MIT.
#-*- coding: utf-8 -*-
import pyodbc # type: ignore
import pyodbc # type: ignore
import logging
from .sql_utils import SqlStatement
from . import sql_utils
from typing import *
from typing import List, Dict, Set, Any, Optional, Callable, Sequence
logger = logging.getLogger(__name__);
logger = logging.getLogger(__name__)
class APplusDBSettings:
"""
Einstellungen, mit welcher DB sich verbunden werden soll.
"""
def __init__(self, server : str, database : str, user : str, password : str):
def __init__(self, server: str, database: str, user: str, password: str):
self.server = server
self.database = database;
self.database = database
self.user = user
self.password = password
def getConnectionString(self) -> str:
"""Liefert den ODBC Connection-String für die Verbindung.
:return: den Connection-String
@ -45,22 +43,23 @@ class APplusDBSettings:
return pyodbc.connect(self.getConnectionString())
def row_to_dict(row : pyodbc.Row) -> Dict[str, Any]:
def row_to_dict(row: pyodbc.Row) -> Dict[str, Any]:
"""Konvertiert eine Zeile in ein Dictionary"""
return dict(zip([t[0] for t in row.cursor_description], row))
def _logSQLWithArgs(sql : SqlStatement, *args : Any) -> None:
def _logSQLWithArgs(sql: SqlStatement, *args: Any) -> None:
if args:
logger.debug("executing '{}' with args {}".format(str(sql), str(args)))
else:
logger.debug("executing '{}'".format(str(sql)))
def rawQueryAll(
cnxn : pyodbc.Connection,
sql : SqlStatement,
*args : Any,
apply : Optional[Callable[[pyodbc.Row], Any]]=None) -> Sequence[Any]:
cnxn: pyodbc.Connection,
sql: SqlStatement,
*args: Any,
apply: Optional[Callable[[pyodbc.Row], Any]] = None) -> Sequence[Any]:
"""
Führt eine SQL Query direkt aus und liefert alle Zeilen zurück.
Wenn apply gesetzt ist, wird die Funktion auf jeder Zeile ausgeführt und das Ergebnis ausgeben, die nicht None sind.
@ -69,44 +68,48 @@ def rawQueryAll(
with cnxn.cursor() as cursor:
cursor.execute(str(sql), *args)
rows = cursor.fetchall();
rows = cursor.fetchall()
if apply is None:
return rows
else:
res = []
for r in rows:
rr = apply(r)
if not (rr == None):
if not (rr is None):
res.append(rr)
return res
def rawQuery(cnxn : pyodbc.Connection, sql : sql_utils.SqlStatement, f : Callable[[pyodbc.Row], None], *args : Any) -> None:
def rawQuery(cnxn: pyodbc.Connection, sql: sql_utils.SqlStatement, f: Callable[[pyodbc.Row], None], *args: Any) -> None:
"""Führt eine SQL Query direkt aus und führt für jede Zeile die übergeben Funktion aus."""
_logSQLWithArgs(sql, *args)
with cnxn.cursor() as cursor:
cursor.execute(str(sql), *args)
for row in cursor:
f(row);
f(row)
def rawQuerySingleRow(cnxn : pyodbc.Connection, sql : SqlStatement, *args : Any) -> Optional[pyodbc.Row]:
def rawQuerySingleRow(cnxn: pyodbc.Connection, sql: SqlStatement, *args: Any) -> Optional[pyodbc.Row]:
"""Führt eine SQL Query direkt aus, die maximal eine Zeile zurückliefern soll. Diese Zeile wird geliefert."""
_logSQLWithArgs(sql, *args)
with cnxn.cursor() as cursor:
cursor.execute(str(sql), *args)
return cursor.fetchone();
return cursor.fetchone()
def rawQuerySingleValue(cnxn : pyodbc.Connection, sql : SqlStatement, *args : Any) -> Any:
def rawQuerySingleValue(cnxn: pyodbc.Connection, sql: SqlStatement, *args: Any) -> Any:
"""Führt eine SQL Query direkt aus, die maximal einen Wert zurückliefern soll. Dieser Wert oder None wird geliefert."""
_logSQLWithArgs(sql, *args)
with cnxn.cursor() as cursor:
cursor.execute(str(sql), *args)
row = cursor.fetchone();
row = cursor.fetchone()
if row:
return row[0];
return row[0]
else:
return None;
return None
def getUniqueFieldsOfTable(cnxn : pyodbc.Connection, table : str) -> Dict[str, List[str]] :
def getUniqueFieldsOfTable(cnxn: pyodbc.Connection, 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.
@ -123,7 +126,7 @@ def getUniqueFieldsOfTable(cnxn : pyodbc.Connection, table : str) -> Dict[str, L
sql.addFields("i.name AS INDEX_NAME", "COL_NAME(ic.OBJECT_ID,ic.column_id) AS COL")
_logSQLWithArgs(sql)
indices : Dict[str, List[str]] = {}
indices: Dict[str, List[str]] = {}
with cnxn.cursor() as cursor:
cursor.execute(str(sql))
for row in cursor:
@ -137,9 +140,9 @@ class DBTableIDs():
"""Klasse, die Mengen von IDs gruppiert nach Tabellen speichert"""
def __init__(self) -> None:
self.data : Dict[str, Set[int]]= {}
self.data: Dict[str, Set[int]] = {}
def add(self, table:str, *ids : int) -> None:
def add(self, table: str, *ids: int) -> None:
"""
fügt Eintrag hinzu
@ -149,11 +152,11 @@ class DBTableIDs():
"""
table = table.upper()
if not (table in self.data):
self.data[table] = set(ids);
self.data[table] = set(ids)
else:
self.data[table].update(ids)
def getTable(self, table : str) -> Set[int]:
def getTable(self, table: str) -> Set[int]:
"""
Liefert die Menge der IDs für eine bestimmte Tabelle.
@ -167,5 +170,3 @@ class DBTableIDs():
def __str__(self) -> str:
return str(self.data)

38
src/PyAPplus64/applus_scripttool.py
View File

@ -6,23 +6,22 @@
# license that can be found in the LICENSE file or at
# https://opensource.org/licenses/MIT.
#-*- coding: utf-8 -*-
from .applus import APplusServer
from . import sql_utils
import lxml.etree as ET # type: ignore
from typing import *
import lxml.etree as ET # type: ignore
from typing import Optional, Tuple, Set
import pathlib
class XMLDefinition:
"""Repräsentation eines XML-Dokuments"""
def __init__(self, root : ET.Element) -> None:
self.root : ET.Element = root
def __init__(self, root: ET.Element) -> None:
self.root: ET.Element = root
"""das Root-Element, repräsentiert "object" aus Datei."""
def __str__(self) -> str:
return ET.tostring(self.root, encoding = "unicode")
return ET.tostring(self.root, encoding="unicode")
def getDuplicate(self) -> Tuple[Set[str], bool]:
"""
@ -31,11 +30,11 @@ class XMLDefinition:
:return: Tuple aus allen Properties und ob dies aus- (True) oder ein-(False) zuschließen sind.
:rtype: Tuple[Set[str], bool]
"""
res : Set[str] = set()
excl = True;
res: Set[str] = set()
excl = True
dupl = self.root.find("duplicate")
if (dupl is None):
return (res, excl);
return (res, excl)
exclS = dupl.get("type", default="exclude")
excl = exclS.casefold() == "exclude"
@ -57,7 +56,7 @@ class APplusScriptTool:
"""
def __init__(self, server : APplusServer) -> None:
def __init__(self, server: APplusServer) -> None:
self.client = server.getClient("p2script", "ScriptTool")
def getCurrentDate(self) -> str:
@ -91,15 +90,15 @@ class APplusScriptTool:
"""
Liefert den Installionspfad des Appservers als PathLib-Path
"""
return pathlib.Path(self.getInstallPath());
return pathlib.Path(self.getInstallPath())
def getInstallPathWebServer(self) -> pathlib.Path:
"""
Liefert den Installionspfad des Webservers als PathLib-Path
"""
return self.getInstallPathAppServer().parents[0].joinpath("WebServer");
return self.getInstallPathAppServer().parents[0].joinpath("WebServer")
def getXMLDefinitionString(self, obj:str, mandant:str="") -> str:
def getXMLDefinitionString(self, obj: str, mandant: str = "") -> str:
"""
Läd die XML-Defintion als String vom APPServer. Auch wenn kein XML-Dokument im Dateisystem gefunden wird,
wird ein String zurückgeliefert, der einen leeren Top-"Object" Knoten enthält. Für gefundene XML-Dokumente
@ -114,7 +113,7 @@ class APplusScriptTool:
"""
return self.client.service.getXMLDefinition2(obj, "")
def getXMLDefinition(self, obj:str, mandant:str="", checkFileExists:bool=False) -> Optional[ET.Element]:
def getXMLDefinition(self, obj: str, mandant: str = "", checkFileExists: bool = False) -> Optional[ET.Element]:
"""
Läd die XML-Definition als String vom APPServer. und parst das XML in ein minidom-Dokument.
@ -127,7 +126,7 @@ class APplusScriptTool:
"""
return ET.fromstring(self.getXMLDefinitionString(obj, mandant=mandant))
def getXMLDefinitionObj(self, obj:str, mandant:str="") -> Optional[XMLDefinition]:
def getXMLDefinitionObj(self, obj: str, mandant: str = "") -> Optional[XMLDefinition]:
"""
Benutzt getXMLDefinitionObj und liefert den Top-Level "Object" Knoten zurück, falls zusätzlich
ein MD5 Knoten existiert, also falls das Dokument wirklich vom Dateisystem geladen werden konnte.
@ -140,19 +139,18 @@ class APplusScriptTool:
:return: das gefundene und geparste XML-Dokument
:rtype: Optional[XMLDefinition]
"""
e = self.getXMLDefinition(obj, mandant=mandant);
e = self.getXMLDefinition(obj, mandant=mandant)
if e is None:
return None
if e.find("md5") is None:
return None;
return None
o = e.find("object")
if o is None:
return None
else:
return XMLDefinition(o);
return XMLDefinition(o)
def getMandant(self) -> str:
"""

35
src/PyAPplus64/applus_server.py
View File

@ -6,9 +6,7 @@
# license that can be found in the LICENSE file or at
# https://opensource.org/licenses/MIT.
#-*- coding: utf-8 -*-
from requests import Session # type: ignore
from requests import Session # type: ignore
from requests.auth import HTTPBasicAuth # type: ignore # or HTTPDigestAuth, or OAuth1, etc.
from zeep import Client
from zeep.transports import Transport
@ -21,23 +19,24 @@ class APplusAppServerSettings:
Einstellungen, mit welchem APplus App-Server sich verbunden werden soll.
"""
def __init__(self, appserver : str, appserverPort : int, user : str, env : Optional[str] = None):
def __init__(self, appserver: str, appserverPort: int, user: str, env: Optional[str] = None):
self.appserver = appserver
self.appserverPort = appserverPort
self.user = user
self.env = env
class APplusWebServerSettings:
"""
Einstellungen, mit welchem APplus Web-Server sich verbunden werden soll.
"""
def __init__(self, baseurl:Optional[str]=None):
self.baseurl : Optional[str] = baseurl;
def __init__(self, baseurl: Optional[str] = None):
self.baseurl: Optional[str] = baseurl
try:
assert (isinstance(self.baseurl, str))
if not (self.baseurl == None) and not (self.baseurl[-1] == "/"):
self.baseurl = self.baseurl + "/";
if not (self.baseurl is None) and not (self.baseurl[-1] == "/"):
self.baseurl = self.baseurl + "/"
except:
pass
@ -48,8 +47,8 @@ class APplusServerConnection:
:param settings: die Einstellungen für die Verbindung mit dem APplus Server
:type settings: APplusAppServerSettings
"""
def __init__(self, settings : APplusAppServerSettings) -> None:
userEnv = settings.user;
def __init__(self, settings: APplusAppServerSettings) -> None:
userEnv = settings.user
if (settings.env):
userEnv += "|" + settings.env
@ -58,11 +57,11 @@ class APplusServerConnection:
self.transport = Transport(cache=SqliteCache(), session=session)
# self.transport = Transport(session=session)
self.clientCache : Dict[str, Client] = {}
self.settings=settings;
self.appserverUrl = "http://" + settings.appserver + ":" + str(settings.appserverPort) + "/";
self.clientCache: Dict[str, Client] = {}
self.settings = settings
self.appserverUrl = "http://" + settings.appserver + ":" + str(settings.appserverPort) + "/"
def getClient(self, package : str, name : str) -> Client:
def getClient(self, package: str, name: str) -> Client:
"""Erzeugt einen zeep - Client.
Mittels dieses Clients kann die WSDL Schnittstelle angesprochen werden.
Wird als *package* "p2core" und als *name* "Table" verwendet und der
@ -77,11 +76,11 @@ class APplusServerConnection:
:return: den Client
:rtype: Client
"""
url = package+"/"+name;
url = package+"/"+name
try:
return self.clientCache[url];
return self.clientCache[url]
except:
fullClientUrl = self.appserverUrl + url + ".jws?wsdl"
client = Client(fullClientUrl, transport=self.transport)
self.clientCache[url] = client;
return client;
self.clientCache[url] = client
return client

44
src/PyAPplus64/applus_sysconf.py
View File

@ -6,9 +6,7 @@
# license that can be found in the LICENSE file or at
# https://opensource.org/licenses/MIT.
#-*- coding: utf-8 -*-
from typing import *
from typing import TYPE_CHECKING, Optional, Dict, Any, Callable, Sequence
if TYPE_CHECKING:
from .applus import APplusServer
@ -23,37 +21,37 @@ class APplusSysConf:
"""
def __init__(self, server : 'APplusServer') -> None:
def __init__(self, server: 'APplusServer') -> None:
self.client = server.getClient("p2system", "SysConf")
self.cache : Dict[str, type] = {}
self.cache: Dict[str, type] = {}
def clearCache(self) -> None:
self.cache = {};
self.cache = {}
def _getGeneral(self, ty:str, f : Callable[[str, str], Any], module:str, name:str, useCache:bool) -> Any:
cacheKey = module + "/" + name + "/" + ty;
def _getGeneral(self, ty: str, f: Callable[[str, str], Any], module: str, name: str, useCache: bool) -> Any:
cacheKey = module + "/" + name + "/" + ty
if useCache and cacheKey in self.cache:
return self.cache[cacheKey]
else:
v = f(module, name);
self.cache[cacheKey] = v;
return v;
v = f(module, name)
self.cache[cacheKey] = v
return v
def getString(self, module:str, name:str, useCache:bool=True) -> str:
return self._getGeneral("string", self.client.service.getString, module, name, useCache);
def getString(self, module: str, name: str, useCache: bool = True) -> str:
return self._getGeneral("string", self.client.service.getString, module, name, useCache)
def getInt(self, module:str, name:str, useCache:bool=True) -> int:
return self._getGeneral("int", self.client.service.getInt, module, name, useCache);
def getInt(self, module: str, name: str, useCache: bool = True) -> int:
return self._getGeneral("int", self.client.service.getInt, module, name, useCache)
def getDouble(self, module:str, name:str, useCache:bool=True) -> float:
return self._getGeneral("double", self.client.service.getDouble, module, name, useCache);
def getDouble(self, module: str, name: str, useCache: bool = True) -> float:
return self._getGeneral("double", self.client.service.getDouble, module, name, useCache)
def getBoolean(self, module:str, name:str, useCache:bool=True) -> bool:
return self._getGeneral("boolean", self.client.service.getBoolean, module, name, useCache);
def getBoolean(self, module: str, name: str, useCache: bool = True) -> bool:
return self._getGeneral("boolean", self.client.service.getBoolean, module, name, useCache)
def getList(self, module : str, name:str, useCache:bool=True, sep:str=",") -> Optional[Sequence[str]]:
s = self.getString(module, name, useCache=useCache);
if (s == None or s == ""):
def getList(self, module: str, name: str, useCache: bool = True, sep: str = ",") -> Optional[Sequence[str]]:
s = self.getString(module, name, useCache=useCache)
if (s is None or s == ""):
return None
return s.split(sep);
return s.split(sep)

100
src/PyAPplus64/applus_usexml.py
View File

@ -6,26 +6,23 @@
# license that can be found in the LICENSE file or at
# https://opensource.org/licenses/MIT.
#-*- coding: utf-8 -*-
import lxml.etree as ET # type: ignore
import lxml.etree as ET # type: ignore
from . import sql_utils
import datetime
from typing import *
from typing import TYPE_CHECKING, Any, Dict, Optional
if TYPE_CHECKING:
from .applus import APplusServer
def _formatValueForXMLRow(v : Any) -> str:
def _formatValueForXMLRow(v: Any) -> str:
"""Hilfsfunktion zum Formatieren eines Wertes für XML"""
if (v is None):
return "";
return ""
if isinstance(v, (int, float)):
return str(v);
return str(v)
elif isinstance(v, str):
return v;
return v
elif isinstance(v, datetime.datetime):
return v.strftime("%Y-%m-%d %H:%M:%S.%f")[:-3]
elif isinstance(v, datetime.date):
@ -60,18 +57,18 @@ class UseXmlRow:
:type cmd: str
"""
def __init__(self, applus : 'APplusServer', table : str, cmd : str) -> None:
def __init__(self, applus: 'APplusServer', table: str, cmd: str) -> None:
self.applus = applus
self.table = table
self.cmd = cmd
self.fields : Dict[str, Any] = {}
self.fields: Dict[str, Any] = {}
def __str__(self) -> str:
return self.toprettyxml()
def _buildXML(self) -> ET.Element :
def _buildXML(self) -> ET.Element:
"""Hilfsfunktion, die das eigentliche XML baut"""
row = ET.Element("row", cmd=self.cmd, table=self.table, nsmap={ "dt" : "urn:schemas-microsoft-com:datatypes"});
row = ET.Element("row", cmd=self.cmd, table=self.table, nsmap={"dt": "urn:schemas-microsoft-com:datatypes"})
for name, value in self.fields.items():
child = ET.Element(name)
@ -80,20 +77,19 @@ class UseXmlRow:
return row
def toprettyxml(self)->str:
def toprettyxml(self) -> str:
"""
Gibt das formatierte XML aus. Dieses kann per useXML an den AppServer übergeben werden.
Dies wird mittels :meth:`exec` automatisiert.
"""
return ET.tostring(self._buildXML(), encoding = "unicode", pretty_print=True)
return ET.tostring(self._buildXML(), encoding="unicode", pretty_print=True)
def getField(self, name:str) -> Any:
def getField(self, name: str) -> Any:
"""Liefert den Wert eines gesetzten Feldes"""
if name is None:
return None
name = sql_utils.normaliseDBfield(name);
name = sql_utils.normaliseDBfield(name)
if name in self.fields:
return self.fields[name]
@ -102,21 +98,21 @@ class UseXmlRow:
else:
return None
def checkFieldSet(self, name:Optional[str]) -> bool:
def checkFieldSet(self, name: Optional[str]) -> bool:
"""Prüft, ob ein Feld gesetzt wurde"""
if name is None:
return False
name = sql_utils.normaliseDBfield(name)
return (name in self.fields) or (name == "MANDANT")
def checkFieldsSet(self, *names : str) -> bool:
def checkFieldsSet(self, *names: str) -> bool:
"""Prüft, ob alle übergebenen Felder gesetzt sind"""
for n in names:
if not (self.checkFieldSet(n)):
return False
return True
def addField(self, name:str|None, value:Any) -> None:
def addField(self, name: Optional[str], value: Any) -> None:
"""
Fügt ein Feld zum Row-Node hinzu.
@ -129,8 +125,7 @@ class UseXmlRow:
self.fields[sql_utils.normaliseDBfield(name)] = value
def addTimestampField(self, id:int, ts:Optional[bytes]=None) -> None:
def addTimestampField(self, id: int, ts: Optional[bytes] = None) -> None:
"""
Fügt ein Timestamp-Feld hinzu. Wird kein Timestamp übergeben, wird mittels der ID der aktuelle
Timestamp aus der DB geladen. Dabei kann ein Fehler auftreten.
@ -146,14 +141,13 @@ class UseXmlRow:
:type ts: bytes
"""
if ts is None:
ts = self.applus.dbQuerySingleValue("select timestamp from " + self.table + " where id = ?", id);
ts = self.applus.dbQuerySingleValue("select timestamp from " + self.table + " where id = ?", id)
if ts:
self.addField("timestamp", ts.hex());
self.addField("timestamp", ts.hex())
else:
raise Exception("kein Eintrag in Tabelle '" + self.table + " mit ID " + str(id) + " gefunden")
def addTimestampIDFields(self, id:int, ts:Optional[bytes]=None) -> None:
def addTimestampIDFields(self, id: int, ts: Optional[bytes] = None) -> None:
"""
Fügt ein Timestamp-Feld sowie ein Feld id hinzu. Wird kein Timestamp übergeben, wird mittels der ID der aktuelle
Timestamp aus der DB geladen. Dabei kann ein Fehler auftreten. Intern wird :meth:`addTimestampField` benutzt.
@ -164,14 +158,14 @@ class UseXmlRow:
:type ts: bytes
"""
self.addField("id", id)
self.addTimestampField(id, ts=ts);
self.addTimestampField(id, ts=ts)
def exec(self) -> Any:
"""
Führt die UseXmlRow mittels useXML aus. Je nach Art der Zeile wird etwas zurückgeliefert oder nicht.
In jedem Fall kann eine Exception geworfen werden.
"""
return self.applus.useXML(self.toprettyxml());
return self.applus.useXML(self.toprettyxml())
class UseXmlRowInsert(UseXmlRow):
@ -184,15 +178,15 @@ class UseXmlRowInsert(UseXmlRow):
:type table: string
"""
def __init__(self, applus:'APplusServer', table:str) -> None:
super().__init__(applus, table, "insert");
def __init__(self, applus: 'APplusServer', table: str) -> None:
super().__init__(applus, table, "insert")
def insert(self) -> int:
"""
Führt das insert aus. Entweder wird dabei eine Exception geworfen oder die ID des neuen Eintrags zurückgegeben.
Dies ist eine Umbenennung von :meth:`exec`.
"""
return super().exec();
return super().exec()
class UseXmlRowDelete(UseXmlRow):
@ -211,17 +205,16 @@ class UseXmlRowDelete(UseXmlRow):
:type ts: bytes optional
"""
def __init__(self, applus:'APplusServer', table:str, id:int, ts:Optional[bytes]=None) -> None:
super().__init__(applus, table, "delete");
self.addTimestampIDFields(id, ts=ts);
def __init__(self, applus: 'APplusServer', table: str, id: int, ts: Optional[bytes] = None) -> None:
super().__init__(applus, table, "delete")
self.addTimestampIDFields(id, ts=ts)
def delete(self) -> None:
"""
Führt das delete aus. Evtl. wird dabei eine Exception geworfen.
Dies ist eine Umbenennung von :meth:`exec`.
"""
super().exec();
super().exec()
class UseXmlRowUpdate(UseXmlRow):
@ -239,18 +232,16 @@ class UseXmlRowUpdate(UseXmlRow):
:type ts: bytes optional
"""
def __init__(self, applus : 'APplusServer', table : str, id : int, ts:Optional[bytes]=None) -> None:
super().__init__(applus, table, "update");
self.addTimestampIDFields(id, ts=ts);
def __init__(self, applus: 'APplusServer', table: str, id: int, ts: Optional[bytes] = None) -> None:
super().__init__(applus, table, "update")
self.addTimestampIDFields(id, ts=ts)
def update(self) -> None:
"""
Führt das update aus. Evtl. wird dabei eine Exception geworfen.
Dies ist eine Umbenennung von :meth:`exec`.
"""
super().exec();
super().exec()
class UseXmlRowInsertOrUpdate(UseXmlRow):
@ -267,21 +258,20 @@ class UseXmlRowInsertOrUpdate(UseXmlRow):
:type table: string
"""
def __init__(self, applus : 'APplusServer', table : str) -> None:
super().__init__(applus, table, "");
def __init__(self, applus: 'APplusServer', table: str) -> None:
super().__init__(applus, table, "")
def checkExists(self) -> int|None:
def checkExists(self) -> Optional[int]:
"""
Prüft, ob der Datensatz bereits in der DB existiert.
Ist dies der Fall, wird die ID geliefert, sonst None
"""
# Baue Bedingung
cond = sql_utils.SqlConditionOr();
cond = sql_utils.SqlConditionOr()
for idx, fs in self.applus.getUniqueFieldsOfTable(self.table).items():
if (self.checkFieldsSet(*fs)):
condIdx = sql_utils.SqlConditionAnd();
condIdx = sql_utils.SqlConditionAnd()
for f in fs:
condIdx.addConditionFieldEq(f, self.getField(f))
cond.addCondition(condIdx)
@ -296,9 +286,9 @@ class UseXmlRowInsertOrUpdate(UseXmlRow):
r = UseXmlRowInsert(self.applus, self.table)
for k, v in self.fields.items():
r.addField(k, v)
return r.insert();
return r.insert()
def update(self, id:Optional[int]=None, ts:Optional[bytes]=None) -> int:
def update(self, id: Optional[int] = None, ts: Optional[bytes] = None) -> int:
"""Führt ein Update aus. Falls ID oder Timestamp nicht übergeben werden, wird
nach einem passenden Objekt gesucht. Existiert das Objekt nicht, wird eine Exception geworfen."""
@ -311,7 +301,7 @@ class UseXmlRowInsertOrUpdate(UseXmlRow):
r = UseXmlRowUpdate(self.applus, self.table, id, ts=ts)
for k, v in self.fields.items():
r.addField(k, v)
r.update();
r.update()
return id
def exec(self) -> int:
@ -320,8 +310,8 @@ class UseXmlRowInsertOrUpdate(UseXmlRow):
der DB existiert. In jedem Fall wird die ID des erzeugten oder geänderten Objekts geliefert.
"""
id = self.checkExists();
if id == None:
id = self.checkExists()
if id is None:
return self.insert()
else:
return self.update(id=id)
@ -332,4 +322,4 @@ class UseXmlRowInsertOrUpdate(UseXmlRow):
Dies ist eine Umbenennung von :meth:`exec`.
Es wird die ID des Eintrages geliefert
"""
return self.exec();
return self.exec()

178
src/PyAPplus64/duplicate.py
View File

@ -6,8 +6,6 @@
# license that can be found in the LICENSE file or at
# https://opensource.org/licenses/MIT.
#-*- coding: utf-8 -*-
"""
Dupliziert ein oder mehrere APplus Business-Objekte
"""
@ -16,18 +14,18 @@ from . import sql_utils
from . import applus_db
from . import applus_usexml
from .applus import APplusServer
import pyodbc # type: ignore
import pyodbc # type: ignore
import traceback
import logging
from typing import *
from typing import List, Set, Optional, Dict, Tuple, Sequence, Any, Union
logger = logging.getLogger(__name__);
logger = logging.getLogger(__name__)
noCopyFields = sql_utils.normaliseDBfieldSet({"INSUSER", "UPDDATE", "TIMESTAMP", "MANDANT", "GUID", "ID", "TIMESTAMP_A", "INSDATE", "ID_A", "UPDUSER"})
"""Menge von Feld-Namen, die nie kopiert werden sollen."""
def getFieldsToCopyForTable(server : APplusServer, table : str, force:bool=True) -> Set[str]:
def getFieldsToCopyForTable(server: APplusServer, table: str, force: bool = True) -> Set[str]:
"""
Bestimmt die für eine Tabelle zu kopierenden Spalten. Dazu wird in den XML-Definitionen geschaut.
Ist dort 'include' hinterlegt, werden diese Spalten verwendet. Ansonsten alle nicht generierten Spalten,
@ -35,10 +33,10 @@ def getFieldsToCopyForTable(server : APplusServer, table : str, force:bool=True)
"""
xmlDefs = server.scripttool.getXMLDefinitionObj(table)
fields : Set[str]
fields: Set[str]
if (xmlDefs is None):
if not force:
raise Exception ("Keine XML-Definitionen für '{}' gefunden".format(table));
raise Exception("Keine XML-Definitionen für '{}' gefunden".format(table))
(fields, excl) = (set(), True)
else:
(fields, excl) = xmlDefs.getDuplicate()
@ -46,8 +44,7 @@ def getFieldsToCopyForTable(server : APplusServer, table : str, force:bool=True)
return fields.difference(noCopyFields)
allFields = server.getTableFields(table, isComputed=False)
return allFields.difference(fields).difference(noCopyFields);
return allFields.difference(fields).difference(noCopyFields)
class FieldsToCopyForTableCache():
@ -55,11 +52,11 @@ class FieldsToCopyForTableCache():
Cache für welche Felder für welche Tabelle kopiert werden sollen
"""
def __init__(self, server : APplusServer) -> None:
def __init__(self, server: APplusServer) -> None:
self.server = server
self.cache : Dict[str, Set[str]]= {}
self.cache: Dict[str, Set[str]] = {}
def getFieldsToCopyForTable(self, table : str) -> Set[str]:
def getFieldsToCopyForTable(self, table: str) -> Set[str]:
"""
Bestimmt die für eine Tabelle zu kopierenden Spalten. Dazu wird in den XML-Definitionen geschaut.
Ist dort 'include' hinterlegt, werden diese Spalten verwendet. Ansonsten alle nicht generierten Spalten,
@ -78,14 +75,14 @@ class FieldsToCopyForTableCache():
return fs
def initFieldsToCopyForTableCacheIfNeeded(server : APplusServer, cache : Optional[FieldsToCopyForTableCache]) -> FieldsToCopyForTableCache:
def initFieldsToCopyForTableCacheIfNeeded(server: APplusServer, cache: Optional[FieldsToCopyForTableCache]) -> FieldsToCopyForTableCache:
"""
Hilfsfunktion, die einen Cache erzeugt, falls dies noch nicht geschehen ist.
"""
if cache is None:
return FieldsToCopyForTableCache(server)
else:
return cache;
return cache
class DuplicateBusinessObject():
@ -101,7 +98,7 @@ class DuplicateBusinessObject():
und geladen werden kann. Ein typisches Beispiel für ein dynamisches Feld ist "GUID".
"""
def __init__(self, table : str, fields : Dict[str, Any], fieldsNotCopied:Dict[str, Any]={}, allowUpdate:bool=False) -> None:
def __init__(self, table: str, fields: Dict[str, Any], fieldsNotCopied: Dict[str, Any] = {}, allowUpdate: bool = False) -> None:
self.table = table
"""für welche Tabelle ist das BusinessObject"""
@ -111,13 +108,13 @@ class DuplicateBusinessObject():
self.fieldsNotCopied = fieldsNotCopied
"""Datenfelder, die im Original vorhanden sind, aber nicht kopiert werden sollen"""
self.dependentObjs : List[Dict[str, Any]] = []
self.dependentObjs: List[Dict[str, Any]] = []
"""Abhängige Objekte"""
self.allowUpdate = allowUpdate
"""Erlaube Updates statt Fehlern, wenn Objekt schon in DB existiert"""
def addDependentBusinessObject(self, dObj : Optional['DuplicateBusinessObject'], *args : Tuple[str, str]) -> None:
def addDependentBusinessObject(self, dObj: Optional['DuplicateBusinessObject'], *args: Tuple[str, str]) -> None:
"""
Fügt ein neues Unterobjekt zum DuplicateBusinessObject hinzu.
Dabei handelt es sich selbst um ein DuplicateBusinessObject, das zusammen mit dem
@ -138,16 +135,16 @@ class DuplicateBusinessObject():
if (dObj is None):
return
args2= {}
args2 = {}
for f1, f2 in args:
args2[sql_utils.normaliseDBfield(f1)] = sql_utils.normaliseDBfield(f2)
self.dependentObjs.append({
"dependentObj" : dObj,
"connection" : args2
"dependentObj": dObj,
"connection": args2
})
def getField(self, field:str, onlyCopied:bool=False) -> Any:
def getField(self, field: str, onlyCopied: bool = False) -> Any:
"""
Schlägt den Wert eines Feldes nach. Wenn onlyCopied gesetzt ist, werden nur Felder zurückgeliefert, die auch kopiert
werden sollen.
@ -162,7 +159,7 @@ class DuplicateBusinessObject():
return None
def insert(self, server : APplusServer) -> applus_db.DBTableIDs:
def insert(self, server: APplusServer) -> applus_db.DBTableIDs:
"""
Fügt alle Objekte zur DB hinzu. Es wird die Menge der IDs der erzeugten
Objekte gruppiert nach Tabellen erzeugt. Falls ein Datensatz schon
@ -172,13 +169,13 @@ class DuplicateBusinessObject():
res = applus_db.DBTableIDs()
def insertDO(do : 'DuplicateBusinessObject') -> Optional[int]:
def insertDO(do: 'DuplicateBusinessObject') -> Optional[int]:
nonlocal res
insertRow : applus_usexml.UseXmlRow
insertRow: applus_usexml.UseXmlRow
if do.allowUpdate:
insertRow = server.mkUseXMLRowInsertOrUpdate(do.table);
insertRow = server.mkUseXMLRowInsertOrUpdate(do.table)
else:
insertRow = server.mkUseXMLRowInsert(do.table);
insertRow = server.mkUseXMLRowInsert(do.table)
for f, v in do.fields.items():
insertRow.addField(f, v)
@ -188,11 +185,11 @@ class DuplicateBusinessObject():
res.add(do.table, id)
return id
except:
msg = traceback.format_exc();
msg = traceback.format_exc()
logger.error("Exception inserting BusinessObjekt: %s\n%s", str(insertRow), msg)
return None
def insertDep(do : 'DuplicateBusinessObject', doID : int, so : 'DuplicateBusinessObject', connect : Dict[str,str]) -> None:
def insertDep(do: 'DuplicateBusinessObject', doID: int, so: 'DuplicateBusinessObject', connect: Dict[str, str]) -> None:
nonlocal res
# Abbruch, wenn do nicht eingefügt wurde
@ -209,7 +206,7 @@ class DuplicateBusinessObject():
# load missing fields from DB
if len(connectMissing) > 0:
sql = sql_utils.SqlStatementSelect(do.table);
sql = sql_utils.SqlStatementSelect(do.table)
sql.where.addConditionFieldEq("id", doID)
for fd in connectMissing:
sql.addFields(fd)
@ -224,8 +221,7 @@ class DuplicateBusinessObject():
if not (id is None):
insertDeps(so, id)
def insertDeps(do : 'DuplicateBusinessObject', doID : int) -> None:
def insertDeps(do: 'DuplicateBusinessObject', doID: int) -> None:
for so in do.dependentObjs:
insertDep(do, doID, so["dependentObj"], so["connection"])
@ -235,8 +231,7 @@ class DuplicateBusinessObject():
return res
def setFields(self, upds : Dict[str, Any]) -> None:
def setFields(self, upds: Dict[str, Any]) -> None:
"""
Setzt Felder des DuplicateBusinessObjektes und falls nötig seiner Unterobjekte.
So kann zum Beispiel die Nummer vor dem Speichern geändert werden.
@ -244,7 +239,7 @@ class DuplicateBusinessObject():
:param upds: Dictionary mit zu setzenden Werten
"""
def setFieldsInternal(dobj : 'DuplicateBusinessObject', upds : Dict[str, Any]) -> None:
def setFieldsInternal(dobj: 'DuplicateBusinessObject', upds: Dict[str, Any]) -> None:
# setze alle Felder des Hauptobjekts
for f, v in upds.items():
dobj.fields[f] = v
@ -257,20 +252,18 @@ class DuplicateBusinessObject():
subupds[fs] = upds[fp]
setFieldsInternal(su["dependentObj"], subupds)
updsNorm : Dict[str, Any] = {}
updsNorm: Dict[str, Any] = {}
for f, v in upds.items():
updsNorm[sql_utils.normaliseDBfield(f)] = v
setFieldsInternal(self, updsNorm)
def _loadDBDuplicateBusinessObjectDict(
server : APplusServer,
table : str,
row : pyodbc.Row,
cache:Optional[FieldsToCopyForTableCache]=None,
allowUpdate:bool=False) -> Optional[DuplicateBusinessObject]:
server: APplusServer,
table: str,
row: pyodbc.Row,
cache: Optional[FieldsToCopyForTableCache] = None,
allowUpdate: bool = False) -> Optional[DuplicateBusinessObject]:
"""
Hilfsfunktion, die ein DuplicateBusinessObjekt erstellt. Die Daten stammen aus
einer PyOdbc Zeile. So ist es möglich, mit nur einem SQL-Statement,
@ -282,7 +275,7 @@ def _loadDBDuplicateBusinessObjectDict(
:param cache: Cache, so dass benötigte Felder nicht immer wieder neu berechnet werden müssen
:return: das neue DuplicateBusinessObject
"""
table = table.upper();
table = table.upper()
def getFieldsToCopy() -> Set[str]:
if cache is None:
@ -290,20 +283,18 @@ def _loadDBDuplicateBusinessObjectDict(
else:
return cache.getFieldsToCopyForTable(table)
def getFields() -> Tuple[Dict[str, Any], Dict[str, Any]]:
ftc = getFieldsToCopy()
fields = {}
fieldsNotCopied = {}
for f, v in applus_db.row_to_dict(row).items():
f = sql_utils.normaliseDBfield(f);
f = sql_utils.normaliseDBfield(f)
if f in ftc:
fields[f] = v
else:
fieldsNotCopied[f] = v
return (fields, fieldsNotCopied)
if (row is None):
return None
@ -312,11 +303,11 @@ def _loadDBDuplicateBusinessObjectDict(
def loadDBDuplicateBusinessObject(
server : APplusServer,
table : str,
cond : sql_utils.SqlCondition,
cache : Optional[FieldsToCopyForTableCache]=None,
allowUpdate : bool = False) -> Optional[DuplicateBusinessObject]:
server: APplusServer,
table: str,
cond: sql_utils.SqlCondition,
cache: Optional[FieldsToCopyForTableCache] = None,
allowUpdate: bool = False) -> Optional[DuplicateBusinessObject]:
"""
Läd ein einzelnes DuplicateBusinessObjekt aus der DB. Die Bedingung sollte dabei
einen eindeutigen Datensatz auswählen. Werden mehrere zurückgeliefert, wird ein
@ -335,23 +326,24 @@ def loadDBDuplicateBusinessObject(
:return: das neue DuplicateBusinessObject
:rtype: Optional[DuplicateBusinessObject]
"""
table = table.upper();
table = table.upper()
def getRow() -> pyodbc.Row:
sql = sql_utils.SqlStatementSelect(table)
sql.setTop(1)
sql.where.addCondition(cond);
sql.where.addCondition(cond)
return server.dbQuerySingleRow(sql)
return _loadDBDuplicateBusinessObjectDict(server, table, getRow(), cache=cache, allowUpdate=allowUpdate);
return _loadDBDuplicateBusinessObjectDict(server, table, getRow(), cache=cache, allowUpdate=allowUpdate)
def loadDBDuplicateBusinessObjectSimpleCond(
server : APplusServer,
table : str,
field : str,
value : Optional[Union[sql_utils.SqlValue, bool]],
cache : Optional[FieldsToCopyForTableCache]=None,
allowUpdate : bool = False) -> Optional[DuplicateBusinessObject]:
server: APplusServer,
table: str,
field: str,
value: Union[sql_utils.SqlValue, bool, None],
cache: Optional[FieldsToCopyForTableCache] = None,
allowUpdate: bool = False) -> Optional[DuplicateBusinessObject]:
"""
Wrapper für loadDBDuplicateBusinessObject, das eine einfache Bedingung benutzt,
bei der ein Feld einen bestimmten Wert haben muss.
@ -373,11 +365,11 @@ def loadDBDuplicateBusinessObjectSimpleCond(
def loadDBDuplicateBusinessObjects(
server : APplusServer,
table : str,
cond : sql_utils.SqlCondition,
cache : Optional[FieldsToCopyForTableCache]=None,
allowUpdate : bool = False) -> Sequence[DuplicateBusinessObject]:
server: APplusServer,
table: str,
cond: sql_utils.SqlCondition,
cache: Optional[FieldsToCopyForTableCache] = None,
allowUpdate: bool = False) -> Sequence[DuplicateBusinessObject]:
"""
Läd eine Liste von DuplicateBusinessObjekten aus der DB. Die Bedingung kann mehrere Datensätze auswählen.
@ -395,20 +387,21 @@ def loadDBDuplicateBusinessObjects(
table = table.upper()
cache = initFieldsToCopyForTableCacheIfNeeded(server, cache)
def processRow(r : pyodbc.Row) -> Optional[DuplicateBusinessObject]:
def processRow(r: pyodbc.Row) -> Optional[DuplicateBusinessObject]:
return _loadDBDuplicateBusinessObjectDict(server, table, r, cache=cache, allowUpdate=allowUpdate)
sql = sql_utils.SqlStatementSelect(table)
sql.where.addCondition(cond)
return server.dbQueryAll(sql, apply=processRow)
def loadDBDuplicateBusinessObjectsSimpleCond(
server : APplusServer,
table : str,
field : str,
value : Optional[Union[sql_utils.SqlValue, bool]],
cache : Optional[FieldsToCopyForTableCache]=None,
allowUpdate : bool = False) -> Sequence[DuplicateBusinessObject]:
server: APplusServer,
table: str,
field: str,
value: Union[sql_utils.SqlValue, bool, None],
cache: Optional[FieldsToCopyForTableCache] = None,
allowUpdate: bool = False) -> Sequence[DuplicateBusinessObject]:
"""
Wrapper für loadDBDuplicateBusinessObjects, das eine einfache Bedingung benutzt,
bei der ein Feld einen bestimmten Wert haben muss.
@ -433,9 +426,9 @@ def loadDBDuplicateBusinessObjectsSimpleCond(
# benutzt werden soll
def loadDBDuplicateAPlan(
server : APplusServer,
aplan : str,
cache:Optional[FieldsToCopyForTableCache]=None) -> Optional[DuplicateBusinessObject]:
server: APplusServer,
aplan: str,
cache: Optional[FieldsToCopyForTableCache] = None) -> Optional[DuplicateBusinessObject]:
"""
Erstelle DuplicateBusinessObject für einzelnen Arbeitsplan.
@ -449,7 +442,7 @@ def loadDBDuplicateAPlan(
:rtype: DuplicateBusinessObject
"""
cache = initFieldsToCopyForTableCacheIfNeeded(server, cache);
cache = initFieldsToCopyForTableCacheIfNeeded(server, cache)
boMain = loadDBDuplicateBusinessObjectSimpleCond(server, "aplan", "APLAN", aplan, cache=cache)
if boMain is None:
return None
@ -460,7 +453,7 @@ def loadDBDuplicateAPlan(
return boMain
def loadDBDuplicateStueli(server : APplusServer, stueli : str, cache:Optional[FieldsToCopyForTableCache]=None) -> Optional[DuplicateBusinessObject]:
def loadDBDuplicateStueli(server: APplusServer, stueli: str, cache: Optional[FieldsToCopyForTableCache] = None) -> Optional[DuplicateBusinessObject]:
"""
Erstelle DuplicateBusinessObject für einzelne Stückliste.
@ -474,7 +467,7 @@ def loadDBDuplicateStueli(server : APplusServer, stueli : str, cache:Optional[Fi
:rtype: Optional[DuplicateBusinessObject]
"""
cache = initFieldsToCopyForTableCacheIfNeeded(server, cache);
cache = initFieldsToCopyForTableCacheIfNeeded(server, cache)
boMain = loadDBDuplicateBusinessObjectSimpleCond(server, "stueli", "stueli", stueli, cache=cache)
if boMain is None:
return None
@ -484,10 +477,11 @@ def loadDBDuplicateStueli(server : APplusServer, stueli : str, cache:Optional[Fi
return boMain
def addSachgruppeDependentObjects(
do : DuplicateBusinessObject,
server : APplusServer,
cache:Optional[FieldsToCopyForTableCache]=None) -> None:
do: DuplicateBusinessObject,
server: APplusServer,
cache: Optional[FieldsToCopyForTableCache] = None) -> None:
"""
Fügt Unterobjekte hinzu, die die Sachgruppenwerte kopieren.
@ -498,9 +492,9 @@ def addSachgruppeDependentObjects(
:type cache: Optional[FieldsToCopyForTableCache]
"""
cache = initFieldsToCopyForTableCacheIfNeeded(server, cache);
cache = initFieldsToCopyForTableCacheIfNeeded(server, cache)
klasse = do.fields.get(sql_utils.normaliseDBfield("SACHGRUPPENKLASSE"), None)
if (klasse == None):
if (klasse is None):
# keine Klasse gesetzt, nichts zu kopieren
return
@ -511,7 +505,7 @@ def addSachgruppeDependentObjects(
sql.where.addConditionFieldEq("tabelle", do.table)
return server.dbQueryAll(sql, apply=lambda r: r.sachgruppe)
gruppen = loadGruppen();
gruppen = loadGruppen()
# Gruppe bearbeiten
def processGruppen() -> None:
@ -525,17 +519,15 @@ def addSachgruppeDependentObjects(
for so in loadDBDuplicateBusinessObjects(server, "sachwert", cond, cache=cache, allowUpdate=True):
do.addDependentBusinessObject(so, ("guid", "instanzguid"))
processGruppen()
def loadDBDuplicateArtikel(
server : APplusServer,
artikel : str,
cache:Optional[FieldsToCopyForTableCache]=None,
dupAplan:bool=True,
dupStueli:bool=True) -> Optional[DuplicateBusinessObject]:
server: APplusServer,
artikel: str,
cache: Optional[FieldsToCopyForTableCache] = None,
dupAplan: bool = True,
dupStueli: bool = True) -> Optional[DuplicateBusinessObject]:
"""
Erstelle DuplicateBusinessObject für einzelnen Artikel.
@ -554,7 +546,7 @@ def loadDBDuplicateArtikel(
:rtype: DuplicateBusinessObject
"""
cache = initFieldsToCopyForTableCacheIfNeeded(server, cache);
cache = initFieldsToCopyForTableCacheIfNeeded(server, cache)
boArt = loadDBDuplicateBusinessObjectSimpleCond(server, "artikel", "ARTIKEL", artikel, cache=cache)
if boArt is None:
return None

63
src/PyAPplus64/pandas.py
View File

@ -8,26 +8,25 @@
"""Pandas Interface für PyAPplus64."""
from typing import Annotated as Ann
import pandas as pd # type: ignore
from pandas._typing import AggFuncType, FilePath, WriteExcelBuffer # type: ignore
import pandas as pd # type: ignore
from pandas._typing import AggFuncType, FilePath, WriteExcelBuffer # type: ignore
import sqlalchemy
import traceback
from .applus import APplusServer
from .applus import sql_utils
from typing import *
from typing import Optional, Callable, Sequence, Tuple, Any, Union
def createSqlAlchemyEngine(server : APplusServer) -> sqlalchemy.Engine:
def createSqlAlchemyEngine(server: APplusServer) -> sqlalchemy.Engine:
"""Erzeugt eine SqlAlchemy-Engine für die Verbindung zur DB."""
return sqlalchemy.create_engine(sqlalchemy.engine.URL.create("mssql+pyodbc", query={"odbc_connect": server.db_settings.getConnectionString()}))
def pandasReadSql(
server : APplusServer,
sql : sql_utils.SqlStatement,
raw:bool=False,
engine:Optional[sqlalchemy.Engine]=None) -> pd.DataFrame:
server: APplusServer,
sql: sql_utils.SqlStatement,
raw: bool = False,
engine: Optional[sqlalchemy.Engine] = None) -> pd.DataFrame:
"""Wrapper für pd.read_sql für sqlalchemy-engine.
:param server: APplusServer für Datenbankverbindung und complete-SQL
@ -36,37 +35,37 @@ def pandasReadSql(
"""
if engine is None:
engine = createSqlAlchemyEngine(server);
engine = createSqlAlchemyEngine(server)
with engine.connect() as conn:
return pd.read_sql(sqlalchemy.text(server.completeSQL(sql, raw=raw)), conn)
def _createHyperLinkGeneral(genOrg : Callable[[], str|int|float], genLink: Callable[[], str]) -> str|int|float:
def _createHyperLinkGeneral(genOrg: Callable[[], Union[str, int, float]], genLink: Callable[[], str]) -> Union[str, int, float]:
"""
Hilfsfunktion zum Generieren eines Excel-Links.
:param genLink: Funktion, die Parameter aufgerufen wird und einen Link generiert
"""
org:str|int|float=""
org2:str|int|float
org: Union[str, int, float] = ""
org2: Union[str, int, float]
try:
org = genOrg();
org = genOrg()
if not org:
return org
else :
else:
if isinstance(org, (int, float)):
org2 = org;
org2 = org
else:
org2 = "\"" + str(org).replace("\"", "\"\"") + "\""
org2 = "\"" + str(org).replace("\"", "\"\"") + "\""
return "=HYPERLINK(\"{}\", {})".format(genLink(), org2)
except:
msg = traceback.format_exc();
print ("Exception: {}".format(msg))
msg = traceback.format_exc()
print("Exception: {}".format(msg))
return org
def mkDataframeColumn(df : pd.DataFrame, makeValue : AggFuncType) -> pd.Series:
def mkDataframeColumn(df: pd.DataFrame, makeValue: AggFuncType) -> pd.Series:
"""
Erzeugt für alle Zeilen eines Dataframes eine neuen Wert. Dies wird benutzt, um eine Spalte zu berechnen.
Diese kann eine Originalspalte ersetzen, oder neu hinzugefügt werden.
@ -74,21 +73,21 @@ def mkDataframeColumn(df : pd.DataFrame, makeValue : AggFuncType) -> pd.Series:
:param df: der Dataframe
:param makeValue: Funktion, die eine Zeile als Parameter bekommt und den neuen Wert berechnet
"""
def mkValueWrapper(r): # type: ignore
def mkValueWrapper(r): # type: ignore
try:
return makeValue(r)
except:
msg = traceback.format_exc();
print ("Exception: {}".format(msg))
msg = traceback.format_exc()
print("Exception: {}".format(msg))
return ""
if (len(df.index) > 0):
return df.apply(mkValueWrapper, axis=1)
else:
return df.apply(lambda r: "", axis=1);
return df.apply(lambda r: "", axis=1)
def mkHyperlinkDataframeColumn(df : pd.DataFrame, makeOrig : AggFuncType, makeLink : Callable[[Any], str]) -> pd.Series :
def mkHyperlinkDataframeColumn(df: pd.DataFrame, makeOrig: AggFuncType, makeLink: Callable[[Any], str]) -> pd.Series:
"""
Erzeugt für alle Zeilen eines Dataframes einen Hyperlink. Dies wird benutzt, um eine Spalte mit einem Hyperlink zu berechnen.
Diese kann eine Originalspalte ersetzen, oder neu hinzugefügt werden.
@ -98,15 +97,15 @@ def mkHyperlinkDataframeColumn(df : pd.DataFrame, makeOrig : AggFuncType, makeLi
:param makeLink: Funktion, die eine Zeile als Parameter bekommt und den Link berechnet
"""
if (len(df.index) > 0):
return df.apply(lambda r: _createHyperLinkGeneral(lambda : makeOrig(r), lambda : makeLink(r)), axis=1)
return df.apply(lambda r: _createHyperLinkGeneral(lambda: makeOrig(r), lambda: makeLink(r)), axis=1)
else:
return df.apply(lambda r: "", axis=1);
return df.apply(lambda r: "", axis=1)
def exportToExcel(
filename:FilePath | WriteExcelBuffer | pd.ExcelWriter,
dfs : Sequence[Tuple[pd.DataFrame, str]],
addTable:bool=True) -> None:
filename: Union[FilePath, WriteExcelBuffer, pd.ExcelWriter],
dfs: Sequence[Tuple[pd.DataFrame, str]],
addTable: bool = True) -> None:
"""
Schreibt eine Menge von Dataframes in eine Excel-Tabelle
@ -126,6 +125,4 @@ def exportToExcel(
ws.add_table(0, 0, max_row, max_col - 1, {'columns': column_settings})
# Spaltenbreiten anpassen
ws.autofit();
ws.autofit()

460
src/PyAPplus64/sql_utils.py
View File

@ -6,7 +6,6 @@
# license that can be found in the LICENSE file or at
# https://opensource.org/licenses/MIT.
#-*- coding: utf-8 -*-
"""
Diese Datei enthält Funktionen für den Bau von SQL Statements, besonders
SELECT-Statements. Es gibt viel ausgefeiltere Methoden für die Erstellung von
@ -22,19 +21,22 @@ APplus. Oft ist es sinnvoll, solche Parameter zu verwenden.
from __future__ import annotations
import datetime
from typing import *
from typing import Set, Sequence, Union, Optional, cast, List
def normaliseDBfield(f : str) -> str:
def normaliseDBfield(f: str) -> str:
"""Normalisiert die Darstellung eines DB-Feldes"""
return str(f).upper();
return str(f).upper()
def normaliseDBfieldSet(s : Set[str]) -> Set[str]:
def normaliseDBfieldSet(s: Set[str]) -> Set[str]:
"""Normalisiert eine Menge von DB-Feldern"""
return {normaliseDBfield(f) for f in s}
def normaliseDBfieldList(l : Sequence[str]) -> Sequence[str]:
def normaliseDBfieldList(fields: Sequence[str]) -> Sequence[str]:
"""Normalisiert eine Menge von DB-Feldern"""
return [normaliseDBfield(f) for f in l]
return [normaliseDBfield(f) for f in fields]
class SqlField():
@ -44,11 +46,12 @@ class SqlField():
:param fn: der Feldname
:type fn: str
"""
def __init__(self, fn : str):
self.field = normaliseDBfield(fn);
def __init__(self, fn: str):
self.field = normaliseDBfield(fn)
def __str__(self) -> str:
return self.field;
return self.field
class SqlFixed():
"""
@ -57,11 +60,12 @@ class SqlFixed():
:param s: der string
:type s: str
"""
def __init__(self, s : str):
self.s = str(s);
def __init__(self, s: str):
self.s = str(s)
def __str__(self) -> str:
return self.s;
return self.s
class SqlDateTime():
"""
@ -70,14 +74,15 @@ class SqlDateTime():
:param dt: der Zeitpunkt
:type dt: Union[datetime.datetime, datetime.date]
"""
def __init__(self, dt:Union[datetime.datetime, datetime.date]=datetime.datetime.now()) -> None:
self.value = dt;
def __init__(self, dt: Union[datetime.datetime, datetime.date] = datetime.datetime.now()) -> None:
self.value = dt
def __str__(self) -> str:
# %f formatiert mit 6 Stellen, also microseconds. Es werden aber nur
# 3 Stellen unterstützt, daher werden 3 weggeworfen.
return self.value.strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3]
class SqlDate():
"""
Wrapper um DateTime, die die Formatierung erleichtern
@ -85,11 +90,12 @@ class SqlDate():
:param d: das Datum
:type d: Union[datetime.datetime, datetime.date]
"""
def __init__(self, d:Union[datetime.datetime, datetime.date]=datetime.datetime.now()) -> None:
self.value = d;
def __init__(self, d: Union[datetime.datetime, datetime.date] = datetime.datetime.now()) -> None:
self.value = d
def __str__(self) -> str:
return self.value.strftime("%Y%m%d");
return self.value.strftime("%Y%m%d")
class SqlTime():
"""
@ -98,12 +104,13 @@ class SqlTime():
:param t: die Zeit
:type t: Union[datetime.datetime, datetime.time]
"""
def __init__(self, t:Union[datetime.datetime, datetime.time]=datetime.datetime.now()) -> None:
def __init__(self, t: Union[datetime.datetime, datetime.time] = datetime.datetime.now()) -> None:
self.value = t
def __str__(self) -> str:
return self.value.strftime("%H:%M:%S.%f")[:-3]
class SqlParam():
"""Hilfsklasse, für einen Parameter (?)"""
def __init__(self) -> None:
@ -112,10 +119,12 @@ class SqlParam():
def __str__(self) -> str:
return "?"
sqlParam = SqlParam()
"""Da SqlParam keinen Zustand hat, reicht ein einzelner statischer Wert"""
def formatSqlValueString(s:str) -> str:
def formatSqlValueString(s: str) -> str:
"""
Formatiert einen String für ein Sql-Statement. Der String wird in "'" eingeschlossen
und Hochkomma im Text maskiert.
@ -127,15 +136,16 @@ def formatSqlValueString(s:str) -> str:
"""
if (s is None):
return "''";
return "''"
return "'" + str(s).replace("'", "''") + "'";
return "'" + str(s).replace("'", "''") + "'"
SqlValue : TypeAlias = Union[str, int, float, SqlParam, SqlField, SqlFixed, SqlDate, SqlDateTime, datetime.datetime, datetime.date, datetime.time]
SqlValue = Union[str, int, float, SqlParam, SqlField, SqlFixed, SqlDate, SqlDateTime, datetime.datetime, datetime.date, datetime.time]
"""Union-Type aller unterstützter SQL-Werte"""
def formatSqlValue(v : SqlValue) -> str:
def formatSqlValue(v: SqlValue) -> str:
"""
Formatiert einen Wert für SQL. Je nachdem um welchen Typ es sich handelt, werden andere Formatierungen verwendet.
@ -145,25 +155,26 @@ def formatSqlValue(v : SqlValue) -> str:
:rtype: str
"""
if (v == None):
raise Exception("formatSqlValue: null not supported");
if v is None:
raise Exception("formatSqlValue: null not supported")
if isinstance(v, (int, float, SqlField)):
return str(v);
return str(v)
elif isinstance(v, str):
return formatSqlValueString(v);
return formatSqlValueString(v)
elif isinstance(v, datetime.datetime):
return "'" + str(SqlDateTime(v)) + "'";
return "'" + str(SqlDateTime(v)) + "'"
elif isinstance(v, datetime.date):
return "'" + str(SqlDate(v)) + "'";
return "'" + str(SqlDate(v)) + "'"
elif isinstance(v, datetime.time):
return "'" + str(SqlTime(v)) + "'";
return "'" + str(SqlTime(v)) + "'"
elif isinstance(v, (SqlDateTime, SqlDate, SqlTime)):
return "'" + str(v) + "'";
return "'" + str(v) + "'"
elif isinstance(v, (SqlParam, SqlFixed)):
return str(v);
return str(v)
else:
raise Exception("formatSqlValue: unsupported type {}".format(type(v)));
raise Exception("formatSqlValue: unsupported type {}".format(type(v)))
class SqlCondition():
"""Eine abstrakte Sql-Bedingung. Unterklassen erledigen die eigentliche Arbeit."""
@ -175,20 +186,21 @@ class SqlCondition():
:return: die Bedingung
:rtype: str
"""
raise Exception("Not implemented");
raise Exception("Not implemented")
def __str__(self) -> str:
return self.getCondition();
return self.getCondition()
class SqlConditionPrepared(SqlCondition):
"""Eine einfache Sql-Bedingung, die immer einen festen String zurückgibt."""
def __init__(self, cond : Union[SqlCondition, str]):
self.cond = str(cond);
def __init__(self, cond: Union[SqlCondition, str]):
self.cond = str(cond)
def getCondition(self) -> str:
return self.cond;
return self.cond
class SqlConditionTrue(SqlConditionPrepared):
"""True-Bedingung"""
@ -196,21 +208,24 @@ class SqlConditionTrue(SqlConditionPrepared):
def __init__(self) -> None:
super().__init__("(1=1)")
class SqlConditionFalse(SqlConditionPrepared):
"""False-Bedingung"""
def __init__(self) -> None:
super().__init__("(1=0)")
class SqlConditionBool(SqlConditionPrepared):
"""Fixe True-oder-False Bedingung"""
def __init__(self, b : bool):
def __init__(self, b: bool):
if b:
super().__init__(SqlConditionTrue())
else:
super().__init__(SqlConditionFalse())
class SqlConditionNot(SqlCondition):
"""
Negation einer anderen Bedingung
@ -219,11 +234,11 @@ class SqlConditionNot(SqlCondition):
:type cond: SqlCondition
"""
def __init__(self, cond : SqlCondition):
self.cond = cond;
def __init__(self, cond: SqlCondition):
self.cond = cond
def getCondition(self) -> str:
return "(not {})".format(self.cond.getCondition());
return "(not {})".format(self.cond.getCondition())
class SqlConditionIsNull(SqlConditionPrepared):
@ -234,13 +249,15 @@ class SqlConditionIsNull(SqlConditionPrepared):
:type v: SqlValue
"""
def __init__(self, v : SqlValue):
def __init__(self, v: SqlValue):
super().__init__("({} is null)".format(formatSqlValue(v)))
class SqlConditionFieldIsNull(SqlConditionIsNull):
def __init__(self, field : str):
def __init__(self, field: str):
super().__init__(SqlField(field))
class SqlConditionIsNotNull(SqlConditionPrepared):
"""
Wert soll nicht null sein
@ -249,13 +266,15 @@ class SqlConditionIsNotNull(SqlConditionPrepared):
:type v: SqlValue
"""
def __init__(self, v : SqlValue):
def __init__(self, v: SqlValue):
super().__init__("({} is not null)".format(formatSqlValue(v)))
class SqlConditionFieldIsNotNull(SqlConditionIsNotNull):
def __init__(self, field : str):
def __init__(self, field: str):
super().__init__(SqlField(field))
class SqlConditionStringStartsWith(SqlConditionPrepared):
"""
Feld soll mit einem bestimmten String beginnen
@ -266,10 +285,10 @@ class SqlConditionStringStartsWith(SqlConditionPrepared):
:type value: str
"""
def __init__(self, field : str, value : str):
cond = "";
def __init__(self, field: str, value: str):
cond = ""
if value:
cond="(left({}, {}) = {})".format(normaliseDBfield(field), len(value), formatSqlValueString(value));
cond = "(left({}, {}) = {})".format(normaliseDBfield(field), len(value), formatSqlValueString(value))
else:
cond = "(1=1)"
super().__init__(cond)
@ -284,9 +303,9 @@ class SqlConditionFieldStringNotEmpty(SqlConditionPrepared):
:type field: str
"""
def __init__(self, field : str):
field = normaliseDBfield(field);
cond="({} is not null and {} != '')".format(field, field);
def __init__(self, field: str):
field = normaliseDBfield(field)
cond = "({} is not null and {} != '')".format(field, field)
super().__init__(cond)
@ -299,10 +318,10 @@ class SqlConditionIn(SqlConditionPrepared):
:param values: die erlaubten Werte
:type values: Sequence[SqlValue]
"""
def __init__(self, value : SqlValue, values : Sequence[SqlValue]):
def __init__(self, value: SqlValue, values: Sequence[SqlValue]):
valuesLen = len(values)
if (valuesLen == 0):
cond : Union[SqlCondition, str] = SqlConditionFalse()
cond: Union[SqlCondition, str] = SqlConditionFalse()
elif (valuesLen == 1):
cond = SqlConditionEq(value, values[0])
else:
@ -312,8 +331,9 @@ class SqlConditionIn(SqlConditionPrepared):
cond = "({} in ({}))".format(formatSqlValue(value), valuesS)
super().__init__(cond)
class SqlConditionFieldIn(SqlConditionIn):
def __init__(self, field:str, values : Sequence[SqlValue]):
def __init__(self, field: str, values: Sequence[SqlValue]):
super().__init__(SqlField(field), values)
@ -324,7 +344,7 @@ class SqlConditionEq(SqlConditionPrepared):
:param value1: der Wert, kann unterschiedliche Typen besitzen
:param value2: der Wert, kann unterschiedliche Typen besitzen
"""
def __init__(self, value1 : Optional[Union[SqlValue, bool]], value2 : Optional[Union[SqlValue, bool]]):
def __init__(self, value1: Union[SqlValue, bool, None], value2: Union[SqlValue, bool, None]):
cond: Union[SqlCondition, str]
if (value1 is None) and (value2 is None):
cond = SqlConditionTrue()
@ -340,13 +360,13 @@ class SqlConditionEq(SqlConditionPrepared):
cond = SqlConditionIsNull(value1)
else:
if isinstance(value1, bool) and isinstance(value2, bool):
cond = SqlConditionBool(value1 == value2);
cond = SqlConditionBool(value1 == value2)
elif isinstance(value1, bool) and not isinstance(value2, bool):
value2 = cast(SqlValue, value2)
if value1:
cond = "({} = 1)".format(formatSqlValue(value2))
else:
cond = "({} = 0 OR {} is null)".format(formatSqlValue(value2), formatSqlValue(value2));
cond = "({} = 0 OR {} is null)".format(formatSqlValue(value2), formatSqlValue(value2))
elif not isinstance(value1, bool) and isinstance(value2, bool):
value1 = cast(SqlValue, value1)
if value2:
@ -356,7 +376,7 @@ class SqlConditionEq(SqlConditionPrepared):
else:
value1 = cast(SqlValue, value1)
value2 = cast(SqlValue, value2)
cond = "({} = {})".format(formatSqlValue(value1), formatSqlValue(value2));
cond = "({} = {})".format(formatSqlValue(value1), formatSqlValue(value2))
super().__init__(cond)
@ -371,11 +391,11 @@ class SqlConditionBinComp(SqlConditionPrepared):
:param value2: der Wert, kann unterschiedliche Typen besitzen
:type value2: SqlValue
"""
def __init__(self, op : str, value1 : SqlValue, value2 : SqlValue):
if not(value1) or not(value2):
def __init__(self, op: str, value1: SqlValue, value2: SqlValue):
if not value1 or not value2:
raise Exception("SqlConditionBinComp: value not provided")
cond = "({} {} {})".format(formatSqlValue(value1), op, formatSqlValue(value2));
cond = "({} {} {})".format(formatSqlValue(value1), op, formatSqlValue(value2))
super().__init__(cond)
@ -386,8 +406,9 @@ class SqlConditionLt(SqlConditionBinComp):
:param value1: der Wert, kann unterschiedliche Typen besitzen
:param value2: der Wert, kann unterschiedliche Typen besitzen
"""
def __init__(self, value1 : SqlValue, value2 : SqlValue):
super().__init__("<", value1, value2)
def __init__(self, value1: SqlValue, value2: SqlValue):
super().__init__("<", value1, value2)
class SqlConditionLe(SqlConditionBinComp):
"""
@ -396,8 +417,9 @@ class SqlConditionLe(SqlConditionBinComp):
:param value1: der Wert, kann unterschiedliche Typen besitzen
:param value2: der Wert, kann unterschiedliche Typen besitzen
"""
def __init__(self, value1 : SqlValue, value2 : SqlValue):
super().__init__("<=", value1, value2)
def __init__(self, value1: SqlValue, value2: SqlValue):
super().__init__("<=", value1, value2)
class SqlConditionGt(SqlConditionBinComp):
"""
@ -406,8 +428,9 @@ class SqlConditionGt(SqlConditionBinComp):
:param value1: der Wert, kann unterschiedliche Typen besitzen
:param value2: der Wert, kann unterschiedliche Typen besitzen
"""
def __init__(self, value1 : SqlValue, value2 : SqlValue):
super().__init__(">", value1, value2)
def __init__(self, value1: SqlValue, value2: SqlValue):
super().__init__(">", value1, value2)
class SqlConditionGe(SqlConditionBinComp):
"""
@ -416,30 +439,35 @@ class SqlConditionGe(SqlConditionBinComp):
:param value1: der Wert, kann unterschiedliche Typen besitzen
:param value2: der Wert, kann unterschiedliche Typen besitzen
"""
def __init__(self, value1 : SqlValue, value2 : SqlValue):
super().__init__(">=", value1, value2)
def __init__(self, value1: SqlValue, value2: SqlValue):
super().__init__(">=", value1, value2)
class SqlConditionFieldEq(SqlConditionEq):
def __init__(self, field : str, value : Optional[Union[SqlValue, bool]]):
def __init__(self, field: str, value: Union[SqlValue, bool, None]):
super().__init__(SqlField(field), value)
class SqlConditionFieldLt(SqlConditionLt):
def __init__(self, field : str, value : SqlValue):
def __init__(self, field: str, value: SqlValue):
super().__init__(SqlField(field), value)
class SqlConditionFieldLe(SqlConditionLe):
def __init__(self, field : str, value : SqlValue):
def __init__(self, field: str, value: SqlValue):
super().__init__(SqlField(field), value)
class SqlConditionFieldGt(SqlConditionGt):
def __init__(self, field : str, value : SqlValue):
def __init__(self, field: str, value: SqlValue):
super().__init__(SqlField(field), value)
class SqlConditionFieldGe(SqlConditionGe):
def __init__(self, field : str, value : SqlValue):
def __init__(self, field: str, value: SqlValue):
super().__init__(SqlField(field), value)
class SqlConditionList(SqlCondition):
"""
Eine SQL Bedingung, die sich aus einer Liste anderer Bedingungen zusammensetzen.
@ -451,94 +479,94 @@ class SqlConditionList(SqlCondition):
:type emptyCond: str
"""
def __init__(self, connector : str, emptyCond : str):
self.connector : str = connector;
self.emptyCond : str = emptyCond;
self.elems : List[SqlCondition] = []
def __init__(self, connector: str, emptyCond: str):
self.connector: str = connector
self.emptyCond: str = emptyCond
self.elems: List[SqlCondition] = []
def addCondition(self, cond : SqlCondition | str | None) -> None:
def addCondition(self, cond: Union[SqlCondition, str, None]) -> None:
if (cond is None):
return
if not (isinstance(cond, SqlCondition)):
cond = SqlConditionPrepared("("+str(cond)+")");
self.elems.append(cond);
cond = SqlConditionPrepared("("+str(cond)+")")
self.elems.append(cond)
def addConditions(self, *conds : SqlCondition | str | None) -> None:
def addConditions(self, *conds: Union[SqlCondition, str, None]) -> None:
for cond in conds:
self.addCondition(cond)
def addConditionFieldStringNotEmpty(self, field : str) -> None:
self.addCondition(SqlConditionFieldStringNotEmpty(field));
def addConditionFieldStringNotEmpty(self, field: str) -> None:
self.addCondition(SqlConditionFieldStringNotEmpty(field))
def addConditionFieldIn(self, field : str, values : Sequence[SqlValue]) -> None:
self.addCondition(SqlConditionFieldIn(field, values));
def addConditionFieldIn(self, field: str, values: Sequence[SqlValue]) -> None:
self.addCondition(SqlConditionFieldIn(field, values))
def addConditionFieldEq(self, field : str, value : Optional[Union[SqlValue, bool]]) -> None:
self.addCondition(SqlConditionFieldEq(field, value));
def addConditionFieldEq(self, field: str, value: Union[SqlValue, bool, None]) -> None:
self.addCondition(SqlConditionFieldEq(field, value))
def addConditionFieldsEq(self, field1 : str, field2 : str) -> None:
self.addCondition(SqlConditionEq(SqlField(field1), SqlField(field2)));
def addConditionFieldsEq(self, field1: str, field2: str) -> None:
self.addCondition(SqlConditionEq(SqlField(field1), SqlField(field2)))
def addConditionEq(self, value1 : Optional[Union[SqlValue, bool]], value2 : Optional[Union[SqlValue, bool]]) -> None:
self.addCondition(SqlConditionEq(value1, value2));
def addConditionEq(self, value1: Union[SqlValue, bool, None], value2: Union[SqlValue, bool, None]) -> None:
self.addCondition(SqlConditionEq(value1, value2))
def addConditionGe(self, value1 : SqlValue, value2 : SqlValue) -> None:
self.addCondition(SqlConditionGe(value1, value2));
def addConditionGe(self, value1: SqlValue, value2: SqlValue) -> None:
self.addCondition(SqlConditionGe(value1, value2))
def addConditionFieldGe(self, field : str, value : SqlValue) -> None:
self.addCondition(SqlConditionGe(SqlField(field), value));
def addConditionFieldGe(self, field: str, value: SqlValue) -> None:
self.addCondition(SqlConditionGe(SqlField(field), value))
def addConditionFieldsGe(self, field1 : str, field2 : str) -> None:
self.addCondition(SqlConditionGe(SqlField(field1), SqlField(field2)));
def addConditionFieldsGe(self, field1: str, field2: str) -> None:
self.addCondition(SqlConditionGe(SqlField(field1), SqlField(field2)))
def addConditionLe(self, value1 : SqlValue, value2 : SqlValue) -> None:
self.addCondition(SqlConditionLe(value1, value2));
def addConditionLe(self, value1: SqlValue, value2: SqlValue) -> None:
self.addCondition(SqlConditionLe(value1, value2))
def addConditionFieldLe(self, field : str, value : SqlValue) -> None:
self.addCondition(SqlConditionLe(SqlField(field), value));
def addConditionFieldLe(self, field: str, value: SqlValue) -> None:
self.addCondition(SqlConditionLe(SqlField(field), value))
def addConditionFieldsLe(self, field1 : str, field2 : str) -> None:
self.addCondition(SqlConditionLe(SqlField(field1), SqlField(field2)));
def addConditionFieldsLe(self, field1: str, field2: str) -> None:
self.addCondition(SqlConditionLe(SqlField(field1), SqlField(field2)))
def addConditionGt(self, value1 : SqlValue, value2 : SqlValue) -> None:
self.addCondition(SqlConditionGt(value1, value2));
def addConditionGt(self, value1: SqlValue, value2: SqlValue) -> None:
self.addCondition(SqlConditionGt(value1, value2))
def addConditionFieldGt(self, field : str, value : SqlValue) -> None:
self.addCondition(SqlConditionGt(SqlField(field), value));
def addConditionFieldGt(self, field: str, value: SqlValue) -> None:
self.addCondition(SqlConditionGt(SqlField(field), value))
def addConditionFieldsGt(self, field1 : str, field2 : str) -> None:
self.addCondition(SqlConditionGt(SqlField(field1), SqlField(field2)));
def addConditionFieldsGt(self, field1: str, field2: str) -> None:
self.addCondition(SqlConditionGt(SqlField(field1), SqlField(field2)))
def addConditionLt(self, value1 : SqlValue, value2 : SqlValue) -> None:
self.addCondition(SqlConditionLt(value1, value2));
def addConditionLt(self, value1: SqlValue, value2: SqlValue) -> None:
self.addCondition(SqlConditionLt(value1, value2))
def addConditionFieldLt(self, field : str, value : SqlValue) -> None:
self.addCondition(SqlConditionLt(SqlField(field), value));
def addConditionFieldLt(self, field: str, value: SqlValue) -> None:
self.addCondition(SqlConditionLt(SqlField(field), value))
def addConditionFieldsLt(self, field1 : str, field2 : str) -> None:
self.addCondition(SqlConditionLt(SqlField(field1), SqlField(field2)));
def addConditionFieldsLt(self, field1: str, field2: str) -> None:
self.addCondition(SqlConditionLt(SqlField(field1), SqlField(field2)))
def addConditionFieldIsNull(self, field : str) -> None:
self.addCondition(SqlConditionFieldIsNull(field));
def addConditionFieldIsNull(self, field: str) -> None:
self.addCondition(SqlConditionFieldIsNull(field))
def addConditionFieldIsNotNull(self, field : str) -> None:
self.addCondition(SqlConditionFieldIsNotNull(field));
def addConditionFieldIsNotNull(self, field: str) -> None:
self.addCondition(SqlConditionFieldIsNotNull(field))
def isEmpty(self) -> bool:
return not(self.elems);
return not self.elems
def getCondition(self) -> str:
match (len(self.elems)):
case 0:
return self.emptyCond;
case 1:
return self.elems[0].getCondition();
case l:
res = "(" + self.elems[0].getCondition();
for i in range(1, l):
res += " {} {}".format(self.connector, self.elems[i].getCondition())
res += ")";
return res;
elemLen = len(self.elems)
if elemLen == 0:
return self.emptyCond
elif elemLen == 1:
return self.elems[0].getCondition()
else:
res = "(" + self.elems[0].getCondition()
for i in range(1, elemLen):
res += " {} {}".format(self.connector, self.elems[i].getCondition())
res += ")"
return res
class SqlConditionDateTimeFieldInRange(SqlConditionPrepared):
@ -550,7 +578,7 @@ class SqlConditionDateTimeFieldInRange(SqlConditionPrepared):
:param datetimeVon: der untere Wert (einschließlich), None erlaubt beliebige Zeiten
:param datetimeBis: der obere Wert (ausschließlich), None erlaubt beliebige Zeiten
"""
def __init__(self, field : str, datetimeVon : datetime.datetime|None, datetimeBis : datetime.datetime|None):
def __init__(self, field: str, datetimeVon: Optional[datetime.datetime], datetimeBis: Optional[datetime.datetime]):
cond = SqlConditionAnd()
if not (datetimeVon is None):
cond.addConditionFieldGe(field, datetimeVon)
@ -558,6 +586,7 @@ class SqlConditionDateTimeFieldInRange(SqlConditionPrepared):
cond.addConditionFieldLt(field, datetimeBis)
super().__init__(str(cond))
class SqlConditionDateTimeFieldInMonth(SqlConditionPrepared):
"""
Liegt Datetime in einem bestimmten Monat?
@ -567,18 +596,20 @@ class SqlConditionDateTimeFieldInMonth(SqlConditionPrepared):
:param year: das Jahr
:param month: der Monat
"""
def __init__(self, field : str, year : int, month : int):
def __init__(self, field: str, year: int, month: int):
if month == 12:
nyear=year+1
nmonth=1;
nyear = year+1
nmonth = 1
else:
nyear=year
nmonth=month+1;
cond = SqlConditionDateTimeFieldInRange(field,
datetime.datetime(year=year, month=month, day=1),
datetime.datetime(year=nyear, month=nmonth, day=1))
nyear = year
nmonth = month+1
cond = SqlConditionDateTimeFieldInRange(
field,
datetime.datetime(year=year, month=month, day=1),
datetime.datetime(year=nyear, month=nmonth, day=1))
super().__init__(str(cond))
class SqlConditionDateTimeFieldInYear(SqlConditionPrepared):
"""
Liegt Datetime in einem bestimmten Jahr?
@ -587,13 +618,15 @@ class SqlConditionDateTimeFieldInYear(SqlConditionPrepared):
:type field: str
:param year: das Jahr
"""
def __init__(self, field : str, year :int) -> None:
nyear=year+1
cond = SqlConditionDateTimeFieldInRange(field,
datetime.datetime(year=year, month=1, day=1),
datetime.datetime(year=nyear, month=1, day=1))
def __init__(self, field: str, year: int) -> None:
nyear = year+1
cond = SqlConditionDateTimeFieldInRange(
field,
datetime.datetime(year=year, month=1, day=1),
datetime.datetime(year=nyear, month=1, day=1))
super().__init__(str(cond))
class SqlConditionDateTimeFieldInDay(SqlConditionPrepared):
"""
Liegt Datetime in einem bestimmten Monat?
@ -604,20 +637,23 @@ class SqlConditionDateTimeFieldInDay(SqlConditionPrepared):
:param month: der Monat
:param day: der Tag
"""
def __init__(self, field : str, year : int, month : int, day : int) -> None:
def __init__(self, field: str, year: int, month: int, day: int) -> None:
d = datetime.datetime(year=year, month=month, day=day)
cond = SqlConditionDateTimeFieldInRange(field,
d,
d + datetime.timedelta(days=1))
cond = SqlConditionDateTimeFieldInRange(
field,
d,
d + datetime.timedelta(days=1))
super().__init__(str(cond))
class SqlConditionAnd(SqlConditionList):
def __init__(self, *conds : Union[SqlCondition, str]) -> None:
def __init__(self, *conds: Union[SqlCondition, str]) -> None:
super().__init__("AND", "(1=1)")
self.addConditions(*conds)
class SqlConditionOr(SqlConditionList):
def __init__(self, *conds : Union[SqlCondition, str]) -> None:
def __init__(self, *conds: Union[SqlCondition, str]) -> None:
super().__init__("OR", "(1=0)")
self.addConditions(*conds)
@ -634,21 +670,21 @@ class SqlJoin():
"""
def __init__(self, joinType : str, table : str, *conds : Union[SqlCondition, str]) -> None:
def __init__(self, joinType: str, table: str, *conds: Union[SqlCondition, str]) -> None:
self.joinType = joinType
self.table = table
self.on : SqlConditionAnd = SqlConditionAnd(*conds)
self.on: SqlConditionAnd = SqlConditionAnd(*conds)
"""Bedingung des Joins, kann noch nachträglich erweitert werden"""
def getJoin(self) -> str:
"""
Liefert den Join als String
"""
return self.joinType + " " + self.table + " ON " + self.on.getCondition();
return self.joinType + " " + self.table + " ON " + self.on.getCondition()
def __str__(self) -> str:
return self.getJoin();
return self.getJoin()
class SqlInnerJoin(SqlJoin):
@ -660,9 +696,10 @@ class SqlInnerJoin(SqlJoin):
:param conds: Bedingungen, die bereits hinzugefügt werden soll. Weitere können über Attribut `on` hinzugefügt werden.
"""
def __init__(self, table : str, *conds : Union[SqlCondition, str]) -> None:
def __init__(self, table: str, *conds: Union[SqlCondition, str]) -> None:
super().__init__("INNER JOIN", table, *conds)
class SqlLeftJoin(SqlJoin):
"""
Ein Left-Join.
@ -671,9 +708,10 @@ class SqlLeftJoin(SqlJoin):
:type table: str
:param conds: Bedingungen, die bereits hinzugefügt werden soll. Weitere können über Attribut `on` hinzugefügt werden.
"""
def __init__(self, table : str, *conds : Union[SqlCondition, str]) -> None:
def __init__(self, table: str, *conds: Union[SqlCondition, str]) -> None:
super().__init__("LEFT JOIN", table, *conds)
class SqlStatementSelect():
"""
Klasse, um einfache Select-Statements zu bauen.
@ -683,71 +721,71 @@ class SqlStatementSelect():
:param fields: kein oder mehrere Felder, die selektiert werden sollen
"""
def __init__(self, table : str, *fields : str) -> None:
self.table : str = table
def __init__(self, table: str, *fields: str) -> None:
self.table: str = table
"""die Tabelle"""
self.top : int = 0
self.top: int = 0
"""wie viele Datensätze auswählen? 0 für alle"""
self.where : SqlConditionList = SqlConditionAnd();
self.where: SqlConditionList = SqlConditionAnd()
"""die Bedingung, Default ist True"""
self.fields : List[str] = []
self.fields: List[str] = []
"""Liste von auszuwählenden Feldern"""
self.addFields(*fields)
self.joins : List[SqlJoin|str] = []
self.joins: List[Union[SqlJoin, str]] = []
"""Joins mit extra Tabellen"""
self.groupBy : List[str] = [];
self.groupBy: List[str] = []
"""die Bedingung, Default ist True"""
self.having : SqlConditionList = SqlConditionAnd();
self.having: SqlConditionList = SqlConditionAnd()
"""die Bedingung having, Default ist True"""
self.order : Optional[str] = None
self.order: Optional[str] = None
"""Sortierung"""
def __str__(self) -> str:
return self.getSql();
return self.getSql()
def addFields(self, *fields : str) -> None:
def addFields(self, *fields: str) -> None:
"""Fügt ein oder mehrere Felder, also auszuwählende Werte zu einem SQL-Statement hinzu."""
for f in fields:
if not (f == None):
if not (f is None):
self.fields.append(f)
def addGroupBy(self, *fields : str) -> None:
def addGroupBy(self, *fields: str) -> None:
"""Fügt ein oder mehrere GroupBy Felder zu einem SQL-Statement hinzu."""
for f in fields:
if not (f == None):
if not (f is None):
self.groupBy.append(f)
def setTop(self, t : int) -> None:
def setTop(self, t: int) -> None:
"""Wie viele Datensätze sollen maximal zurückgeliefert werden? 0 für alle"""
self.top = t
def addFieldsTable(self, table : str, *fields : str) -> None:
def addFieldsTable(self, table: str, *fields: str) -> None:
"""
Fügt ein oder mehrere Felder, die zu einer Tabelle gehören zu einem SQL-Statement hinzu.
Felder sind Strings. Vor jeden dieser Strings wird die Tabelle mit einem Punkt getrennt gesetzt.
Dies kann im Vergleich zu 'addFields' Schreibarbeit erleitern.
"""
for f in fields:
if not (f == None):
if not (f is None):
self.fields.append(table + "." + str(f))
def addJoin(self, j : SqlJoin|str) -> None:
def addJoin(self, j: Union[SqlJoin, str]) -> None:
"""Fügt ein Join zum SQL-Statement hinzu. Beispiel: 'LEFT JOIN personal p ON t.UPDUSER = p.PERSONAL'"""
self.joins.append(j)
def addLeftJoin(self, table : str, *conds : Union[SqlCondition, str]) -> SqlLeftJoin:
def addLeftJoin(self, table: str, *conds: Union[SqlCondition, str]) -> SqlLeftJoin:
j = SqlLeftJoin(table, *conds)
self.addJoin(j)
return j
def addInnerJoin(self, table : str, *conds : Union[SqlCondition, str]) -> SqlInnerJoin:
def addInnerJoin(self, table: str, *conds: Union[SqlCondition, str]) -> SqlInnerJoin:
j = SqlInnerJoin(table, *conds)
self.addJoin(j)
return j
@ -755,38 +793,37 @@ class SqlStatementSelect():
def getSql(self) -> str:
"""Liefert das SQL-SELECT-Statement als String"""
def getFields() -> str:
match (len(self.fields)):
case 0:
return "*";
case 1:
return str(self.fields[0]);
case l:
res = str(self.fields[0]);
for i in range(1, l):
res += ", " + str(self.fields[i]);
return res;
fieldsLen = len(self.fields)
if fieldsLen == 0:
return "*"
elif fieldsLen == 1:
return str(self.fields[0])
else:
res = str(self.fields[0])
for i in range(1, fieldsLen):
res += ", " + str(self.fields[i])
return res
def getGroupBy() -> str:
match (len(self.groupBy)):
case 0:
return "";
case l:
res = " GROUP BY " + str(self.fields[0])
for i in range(1, l):
res += ", " + str(self.fields[i]);
if not (self.having.isEmpty()):
res += " HAVING " + str(self.having)
return res;
groupByLen = len(self.groupBy)
if groupByLen == 0:
return ""
else:
res = " GROUP BY " + str(self.fields[0])
for i in range(1, groupByLen):
res += ", " + str(self.fields[i])
if not (self.having.isEmpty()):
res += " HAVING " + str(self.having)
return res
def getJoins() -> str:
match (len(self.joins)):
case 0:
return ""
case l:
res = "";
for i in range(0, l):
res += " " + str(self.joins[i])
return res
if (len(self.joins) == 0):
return ""
else:
res = ""
for i in range(0, len(self.joins)):
res += " " + str(self.joins[i])
return res
def getWhere() -> str:
if self.where.isEmpty():
@ -795,7 +832,7 @@ class SqlStatementSelect():
return " WHERE " + str(self.where)
def getOrder() -> str:
if self.order == None:
if self.order is None:
return ""
else:
return " ORDER BY " + str(self.order)
@ -806,8 +843,7 @@ class SqlStatementSelect():
else:
return "TOP " + str(self.top) + " "
return "SELECT " + getTop() + getFields() + " FROM " + self.table + getJoins() + getWhere() + getGroupBy() + getOrder();
return "SELECT " + getTop() + getFields() + " FROM " + self.table + getJoins() + getWhere() + getGroupBy() + getOrder()
SqlStatement : TypeAlias = Union [SqlStatementSelect, str]
SqlStatement = Union[SqlStatementSelect, str]

24
src/PyAPplus64/utils.py
View File

@ -6,13 +6,12 @@
# license that can be found in the LICENSE file or at
# https://opensource.org/licenses/MIT.
#-*- coding: utf-8 -*-
import pathlib
import datetime
from typing import *
from typing import Set, Union
def checkDirExists(dir : Union[str, pathlib.Path]) -> pathlib.Path:
def checkDirExists(dir: Union[str, pathlib.Path]) -> pathlib.Path:
"""Prüft, ob ein Verzeichnis existiert. Ist dies nicht möglich, wird eine Exception geworfen.
:param dir: das Verzeichnis
@ -26,19 +25,19 @@ def checkDirExists(dir : Union[str, pathlib.Path]) -> pathlib.Path:
dir = dir.resolve()
if not (dir.exists()):
raise Exception("Verzeichnis '" + str(dir) + "' nicht gefunden");
raise Exception("Verzeichnis '" + str(dir) + "' nicht gefunden")
if not (dir.is_dir()):
raise Exception("'" + str(dir) + "' ist kein Verzeichnis");
return dir;
raise Exception("'" + str(dir) + "' ist kein Verzeichnis")
return dir
def formatDateTimeForAPplus(v : Union[datetime.datetime, datetime.date, datetime.time]) -> str:
def formatDateTimeForAPplus(v: Union[datetime.datetime, datetime.date, datetime.time]) -> str:
"""Formatiert ein Datum oder eine Uhrzeit für APplus"""
if (v == None):
return "";
if v is None:
return ""
elif isinstance(v, str):
return v;
return v
elif isinstance(v, datetime.datetime):
return v.strftime("%Y-%m-%d %H:%M:%S.%f")[:-3]
elif isinstance(v, datetime.date):
@ -48,7 +47,8 @@ def formatDateTimeForAPplus(v : Union[datetime.datetime, datetime.date, datetime
else:
return str(v)
def containsOnlyAllowedChars(charset : Set[str], s : str) -> bool:
def containsOnlyAllowedChars(charset: Set[str], s: str) -> bool:
"""Enthält ein String nur erlaubte Zeichen?"""
for c in s:
if not (c in charset):

10
tests/test_applus_db.py
View File

@ -7,17 +7,17 @@
# https://opensource.org/licenses/MIT.
from PyAPplus64 import applus_db
import datetime
def test_DBTableIDs1() -> None:
ids = applus_db.DBTableIDs();
ids = applus_db.DBTableIDs()
assert (str(ids) == "{}")
ids.add("t1", 1)
assert (str(ids) == "{'T1': {1}}")
ids.add("t1", 2,3,4)
ids.add("t1", 2, 3, 4)
assert (str(ids) == "{'T1': {1, 2, 3, 4}}")
assert (ids.getTable("T1") == {1, 2, 3, 4})
assert (ids.getTable("T2") == set())
ids.add("t2", 2,3,4)
assert (ids.getTable("T2") == {2,3,4})
ids.add("t2", 2, 3, 4)
assert (ids.getTable("T2") == {2, 3, 4})
assert (str(ids) == "{'T1': {1, 2, 3, 4}, 'T2': {2, 3, 4}}")

239
tests/test_sql_utils.py
View File

@ -9,284 +9,350 @@
from PyAPplus64 import sql_utils
import datetime
def test_normaliseDBField1() -> None:
assert (sql_utils.normaliseDBfield("aAa") == "AAA")
assert (sql_utils.normaliseDBfield("a#Aa") == "A#AA")
assert (sql_utils.normaliseDBfield("2") == "2")
def test_normaliseDBFieldSet() -> None:
assert (sql_utils.normaliseDBfieldSet(set()) == set())
assert (sql_utils.normaliseDBfieldSet({"aAa", "b", "c", "2"}) == {"2", "AAA", "B", "C"})
def test_normaliseDBFieldList() -> None:
assert (sql_utils.normaliseDBfieldList([]) == [])
assert (sql_utils.normaliseDBfieldList(["aAa", "b", "c", "2"]) == ["AAA", "B", "C", "2"])
def test_SqlField1() -> None:
assert (str(sql_utils.SqlField("abc")) == "ABC")
def test_SqlField2() -> None:
assert (str(sql_utils.SqlField("t.abc")) == "T.ABC")
def test_SqlParam() -> None:
assert (str(sql_utils.sqlParam) == "?")
def test_SqlDateTime() -> None:
dt = datetime.datetime(year=2023, month=1, day=12, hour=9, minute=59, second=12, microsecond=2344)
assert (str(sql_utils.SqlDateTime(dt)) == "2023-01-12T09:59:12.002")
def test_SqlDate() -> None:
dt = datetime.datetime(year=2023, month=1, day=12, hour=9, minute=59, second=12, microsecond=2344)
assert (str(sql_utils.SqlDate(dt)) == "20230112")
def test_formatSqlValueString1() -> None:
assert(sql_utils.formatSqlValueString("") == "''");
assert (sql_utils.formatSqlValueString("") == "''")
def test_formatSqlValueString2() -> None:
assert(sql_utils.formatSqlValueString("abc") == "'abc'");
assert (sql_utils.formatSqlValueString("abc") == "'abc'")
def test_formatSqlValueString3() -> None:
assert(sql_utils.formatSqlValueString("a b c") == "'a b c'");
assert (sql_utils.formatSqlValueString("a b c") == "'a b c'")
def test_formatSqlValueString4() -> None:
assert(sql_utils.formatSqlValueString("a \"b\" c") == "'a \"b\" c'");
assert (sql_utils.formatSqlValueString("a \"b\" c") == "'a \"b\" c'")
def test_formatSqlValueString5() -> None:
assert(sql_utils.formatSqlValueString("a 'b'\nc") == "'a ''b''\nc'");
assert (sql_utils.formatSqlValueString("a 'b'\nc") == "'a ''b''\nc'")
def test_formatSqlValue1() -> None:
assert(sql_utils.formatSqlValue(2) == "2");
assert (sql_utils.formatSqlValue(2) == "2")
def test_formatSqlValue2() -> None:
assert(sql_utils.formatSqlValue(2.4) == "2.4");
assert (sql_utils.formatSqlValue(2.4) == "2.4")
def test_formatSqlValue3() -> None:
assert(sql_utils.formatSqlValue("AA") == "'AA'");
assert (sql_utils.formatSqlValue("AA") == "'AA'")
def test_formatSqlValue4() -> None:
assert(sql_utils.formatSqlValue(sql_utils.SqlField("aa")) == "AA");
assert (sql_utils.formatSqlValue(sql_utils.SqlField("aa")) == "AA")
def test_formatSqlValue5() -> None:
assert(sql_utils.formatSqlValue(0) == "0");
assert (sql_utils.formatSqlValue(0) == "0")
def test_formatSqlValue6() -> None:
dt = datetime.datetime(year=2023, month=1, day=12, hour=9, minute=59, second=12, microsecond=2344)
assert(sql_utils.formatSqlValue(sql_utils.SqlDateTime(dt)) == "'2023-01-12T09:59:12.002'");
assert (sql_utils.formatSqlValue(sql_utils.SqlDateTime(dt)) == "'2023-01-12T09:59:12.002'")
def test_SqlConditionTrue() -> None:
assert(str(sql_utils.SqlConditionTrue()) == "(1=1)");
assert (str(sql_utils.SqlConditionTrue()) == "(1=1)")
def test_SqlConditionFalse() -> None:
assert(str(sql_utils.SqlConditionFalse()) == "(1=0)");
assert (str(sql_utils.SqlConditionFalse()) == "(1=0)")
def test_SqlConditionBool1() -> None:
assert(str(sql_utils.SqlConditionBool(True)) == "(1=1)");
assert (str(sql_utils.SqlConditionBool(True)) == "(1=1)")
def test_SqlConditionBool2() -> None:
assert(str(sql_utils.SqlConditionBool(False)) == "(1=0)");
assert (str(sql_utils.SqlConditionBool(False)) == "(1=0)")
def test_SqlConditionIsNull() -> None:
cond = sql_utils.SqlConditionIsNull("AA");
assert(str(cond) == "('AA' is null)");
cond = sql_utils.SqlConditionIsNull("AA")
assert (str(cond) == "('AA' is null)")
def test_SqlConditionIsNotNull() -> None:
cond = sql_utils.SqlConditionIsNotNull("AA");
assert(str(cond) == "('AA' is not null)");
cond = sql_utils.SqlConditionIsNotNull("AA")
assert (str(cond) == "('AA' is not null)")
def test_SqlConditionNot() -> None:
cond1 = sql_utils.SqlConditionIsNull("AA");
cond = sql_utils.SqlConditionNot(cond1);
assert(str(cond) == "(not ('AA' is null))");
cond1 = sql_utils.SqlConditionIsNull("AA")
cond = sql_utils.SqlConditionNot(cond1)
assert (str(cond) == "(not ('AA' is null))")
def test_SqlConditionStringStartsWith() -> None:
cond = sql_utils.SqlConditionStringStartsWith("f", "a'an")
assert(str(cond) == "(left(F, 4) = 'a''an')");
assert (str(cond) == "(left(F, 4) = 'a''an')")
def test_SqlConditionIn1() -> None:
cond = sql_utils.SqlConditionIn(sql_utils.SqlField("f"), [])
assert(str(cond) == "(1=0)");
assert (str(cond) == "(1=0)")
def test_SqlConditionIn2() -> None:
cond = sql_utils.SqlConditionIn(sql_utils.SqlField("f"), ["a"])
assert(str(cond) == "(F = 'a')");
assert (str(cond) == "(F = 'a')")
def test_SqlConditionIn3() -> None:
cond = sql_utils.SqlConditionIn(sql_utils.SqlField("f"), ["a", "a'A", "b", "c"])
assert(str(cond) == "(F in ('a', 'a''A', 'b', 'c'))");
assert (str(cond) == "(F in ('a', 'a''A', 'b', 'c'))")
def test_SqlConditionStringNotEmpty1() -> None:
cond = sql_utils.SqlConditionFieldStringNotEmpty("f")
assert(str(cond) == "(F is not null and F != '')");
assert (str(cond) == "(F is not null and F != '')")
def test_SqlConditionEq1() -> None:
cond = sql_utils.SqlConditionEq("f1", None)
assert(str(cond) == "('f1' is null)");
assert (str(cond) == "('f1' is null)")
def test_SqlConditionEq2() -> None:
cond = sql_utils.SqlConditionEq(None, "f1")
assert(str(cond) == "('f1' is null)");
assert (str(cond) == "('f1' is null)")
def test_SqlConditionEq3() -> None:
cond = sql_utils.SqlConditionEq(sql_utils.SqlField("f1"), sql_utils.SqlField("f2"))
assert(str(cond) == "(F1 = F2)");
assert (str(cond) == "(F1 = F2)")
def test_SqlConditionEq4() -> None:
cond = sql_utils.SqlConditionEq(sql_utils.SqlField("f1"), "aa'a")
assert(str(cond) == "(F1 = 'aa''a')");
assert (str(cond) == "(F1 = 'aa''a')")
def test_SqlConditionEq5() -> None:
cond = sql_utils.SqlConditionEq(sql_utils.SqlField("f1"), 2)
assert(str(cond) == "(F1 = 2)");
assert (str(cond) == "(F1 = 2)")
def test_SqlConditionEq6() -> None:
cond = sql_utils.SqlConditionEq(sql_utils.SqlField("f1"), True)
assert(str(cond) == "(F1 = 1)");
assert (str(cond) == "(F1 = 1)")
def test_SqlConditionEq7() -> None:
cond = sql_utils.SqlConditionEq(sql_utils.SqlField("f1"), False)
assert(str(cond) == "(F1 = 0 OR F1 is null)");
assert (str(cond) == "(F1 = 0 OR F1 is null)")
def test_SqlConditionEq8() -> None:
cond = sql_utils.SqlConditionEq(True, sql_utils.SqlField("f1"))
assert(str(cond) == "(F1 = 1)");
assert (str(cond) == "(F1 = 1)")
def test_SqlConditionEq9() -> None:
cond = sql_utils.SqlConditionEq(False, sql_utils.SqlField("f1"))
assert(str(cond) == "(F1 = 0 OR F1 is null)");
assert (str(cond) == "(F1 = 0 OR F1 is null)")
def test_SqlConditionEq10() -> None:
cond = sql_utils.SqlConditionEq(False, True)
assert(str(cond) == "(1=0)");
assert (str(cond) == "(1=0)")
def test_SqlConditionEq11() -> None:
cond = sql_utils.SqlConditionEq(True, True)
assert(str(cond) == "(1=1)");
assert (str(cond) == "(1=1)")
def test_SqlConditionFieldEq1() -> None:
cond = sql_utils.SqlConditionFieldEq("f1", None)
assert(str(cond) == "(F1 is null)");
assert (str(cond) == "(F1 is null)")
def test_SqlConditionFieldEq2() -> None:
cond = sql_utils.SqlConditionFieldEq("f1", sql_utils.SqlField("f2"))
assert(str(cond) == "(F1 = F2)");
assert (str(cond) == "(F1 = F2)")
def test_SqlConditionFieldEq3() -> None:
cond = sql_utils.SqlConditionFieldEq("f1", "aa'a")
assert(str(cond) == "(F1 = 'aa''a')");
assert (str(cond) == "(F1 = 'aa''a')")
def test_SqlConditionFieldEq4() -> None:
cond = sql_utils.SqlConditionFieldEq("f1", 2)
assert(str(cond) == "(F1 = 2)");
assert (str(cond) == "(F1 = 2)")
def test_SqlConditionFieldEq5() -> None:
cond = sql_utils.SqlConditionFieldEq("f1", sql_utils.sqlParam)
assert(str(cond) == "(F1 = ?)");
assert (str(cond) == "(F1 = ?)")
def test_SqlConditionLt1() -> None:
cond = sql_utils.SqlConditionLt(sql_utils.SqlField("f"), sql_utils.SqlDate(datetime.date(year=2022, month=12, day=12)))
assert(str(cond) == "(F < '20221212')");
assert (str(cond) == "(F < '20221212')")
def test_SqlConditionLt2() -> None:
cond = sql_utils.SqlConditionLt(2, sql_utils.SqlField("f"))
assert(str(cond) == "(2 < F)");
assert (str(cond) == "(2 < F)")
def test_SqlConditionGt1() -> None:
cond = sql_utils.SqlConditionGt(sql_utils.SqlField("f"), sql_utils.SqlDate(datetime.date(year=2022, month=12, day=12)))
assert(str(cond) == "(F > '20221212')");
assert (str(cond) == "(F > '20221212')")
def test_SqlConditionGt2() -> None:
cond = sql_utils.SqlConditionGt(2, sql_utils.SqlField("f"))
assert(str(cond) == "(2 > F)");
assert (str(cond) == "(2 > F)")
def test_SqlConditionLe1() -> None:
cond = sql_utils.SqlConditionLe(sql_utils.SqlField("f"), sql_utils.SqlDate(datetime.date(year=2022, month=12, day=12)))
assert(str(cond) == "(F <= '20221212')");
assert (str(cond) == "(F <= '20221212')")
def test_SqlConditionLe2() -> None:
cond = sql_utils.SqlConditionLe(2, sql_utils.SqlField("f"))
assert(str(cond) == "(2 <= F)");
assert (str(cond) == "(2 <= F)")
def test_SqlConditionGe1() -> None:
cond = sql_utils.SqlConditionGe(sql_utils.SqlField("f"), sql_utils.SqlDate(datetime.date(year=2022, month=12, day=12)))
assert(str(cond) == "(F >= '20221212')");
assert (str(cond) == "(F >= '20221212')")
def test_SqlConditionGe2() -> None:
cond = sql_utils.SqlConditionGe(2, sql_utils.SqlField("f"))
assert(str(cond) == "(2 >= F)");
assert (str(cond) == "(2 >= F)")
def test_SqlConditionFieldLt1() -> None:
cond = sql_utils.SqlConditionFieldLt("f", sql_utils.SqlDate(datetime.date(year=2022, month=12, day=12)))
assert(str(cond) == "(F < '20221212')");
assert (str(cond) == "(F < '20221212')")
def test_SqlConditionFieldLe1() -> None:
cond = sql_utils.SqlConditionFieldLe("f", sql_utils.SqlDate(datetime.date(year=2022, month=12, day=12)))
assert(str(cond) == "(F <= '20221212')");
assert (str(cond) == "(F <= '20221212')")
def test_SqlConditionFieldGt1() -> None:
cond = sql_utils.SqlConditionFieldGt("f", sql_utils.SqlDate(datetime.date(year=2022, month=12, day=12)))
assert(str(cond) == "(F > '20221212')");
assert (str(cond) == "(F > '20221212')")
def test_SqlConditionFieldGe1() -> None:
cond = sql_utils.SqlConditionFieldGe("f", sql_utils.SqlDate(datetime.date(year=2022, month=12, day=12)))
assert(str(cond) == "(F >= '20221212')");
assert (str(cond) == "(F >= '20221212')")
def test_SqlConditionAnd1() -> None:
conj = sql_utils.SqlConditionAnd();
assert(str(conj) == "(1=1)");
conj = sql_utils.SqlConditionAnd()
assert (str(conj) == "(1=1)")
def test_SqlConditionAnd2() -> None:
cond1 = sql_utils.SqlConditionPrepared("cond1");
conj = sql_utils.SqlConditionAnd();
cond1 = sql_utils.SqlConditionPrepared("cond1")
conj = sql_utils.SqlConditionAnd()
conj.addCondition(cond1)
assert(str(conj) == "cond1");
assert (str(conj) == "cond1")
def test_SqlConditionAnd3() -> None:
cond1 = sql_utils.SqlConditionPrepared("cond1");
cond2 = sql_utils.SqlConditionPrepared("cond2");
conj = sql_utils.SqlConditionAnd();
cond1 = sql_utils.SqlConditionPrepared("cond1")
cond2 = sql_utils.SqlConditionPrepared("cond2")
conj = sql_utils.SqlConditionAnd()
conj.addCondition(cond1)
conj.addCondition(cond2)
assert(str(conj) == "(cond1 AND cond2)");
assert (str(conj) == "(cond1 AND cond2)")
def test_SqlConditionAnd4() -> None:
cond1 = sql_utils.SqlConditionPrepared("cond1");
cond2 = sql_utils.SqlConditionPrepared("cond2");
cond3 = sql_utils.SqlConditionPrepared("cond3");
conj = sql_utils.SqlConditionAnd();
cond1 = sql_utils.SqlConditionPrepared("cond1")
cond2 = sql_utils.SqlConditionPrepared("cond2")
cond3 = sql_utils.SqlConditionPrepared("cond3")
conj = sql_utils.SqlConditionAnd()
conj.addCondition(cond1)
conj.addCondition(cond2)
conj.addCondition(cond3)
assert(str(conj) == "(cond1 AND cond2 AND cond3)");
assert (str(conj) == "(cond1 AND cond2 AND cond3)")
def test_SqlConditionOr1() -> None:
conj = sql_utils.SqlConditionOr();
assert(str(conj) == "(1=0)");
conj = sql_utils.SqlConditionOr()
assert (str(conj) == "(1=0)")
def test_SqlConditionOr2() -> None:
cond1 = sql_utils.SqlConditionPrepared("cond1");
conj = sql_utils.SqlConditionOr();
cond1 = sql_utils.SqlConditionPrepared("cond1")
conj = sql_utils.SqlConditionOr()
conj.addCondition(cond1)
assert(str(conj) == "cond1");
assert (str(conj) == "cond1")
def test_SqlConditionOr3() -> None:
cond1 = sql_utils.SqlConditionPrepared("cond1");
cond2 = sql_utils.SqlConditionPrepared("cond2");
conj = sql_utils.SqlConditionOr();
cond1 = sql_utils.SqlConditionPrepared("cond1")
cond2 = sql_utils.SqlConditionPrepared("cond2")
conj = sql_utils.SqlConditionOr()
conj.addCondition(cond1)
conj.addCondition(cond2)
assert(str(conj) == "(cond1 OR cond2)");
assert (str(conj) == "(cond1 OR cond2)")
def test_SqlConditionOr4() -> None:
cond1 = sql_utils.SqlConditionPrepared("cond1");
cond2 = sql_utils.SqlConditionPrepared("cond2");
cond3 = sql_utils.SqlConditionPrepared("cond3");
conj = sql_utils.SqlConditionOr();
cond1 = sql_utils.SqlConditionPrepared("cond1")
cond2 = sql_utils.SqlConditionPrepared("cond2")
cond3 = sql_utils.SqlConditionPrepared("cond3")
conj = sql_utils.SqlConditionOr()
conj.addCondition(cond1)
conj.addCondition(cond2)
conj.addCondition(cond3)
assert(str(conj) == "(cond1 OR cond2 OR cond3)");
assert (str(conj) == "(cond1 OR cond2 OR cond3)")
def test_SqlStatementSelect1() -> None:
sql = sql_utils.SqlStatementSelect("tabelle t")
@ -323,6 +389,7 @@ def test_SqlStatementSelect2() -> None:
sql.addJoin("left join t3 on cond3")
assert (str(sql) == "SELECT * FROM t1 left join t2 on cond2 left join t3 on cond3")
def test_SqlStatementSelect4() -> None:
sql = sql_utils.SqlStatementSelect("t")
sql.where.addCondition("cond1")
@ -331,9 +398,10 @@ def test_SqlStatementSelect4() -> None:
sql.where.addCondition("cond2")
assert (str(sql) == "SELECT * FROM t WHERE ((cond1) AND (cond2))")
def test_SqlStatementSelect5() -> None:
sql = sql_utils.SqlStatementSelect("t")
cond = sql_utils.SqlConditionOr();
cond = sql_utils.SqlConditionOr()
sql.where.addCondition(cond)
cond.addCondition("cond1")
assert (str(sql) == "SELECT * FROM t WHERE (cond1)")
@ -341,9 +409,10 @@ def test_SqlStatementSelect5() -> None:
cond.addCondition("cond2")
assert (str(sql) == "SELECT * FROM t WHERE ((cond1) OR (cond2))")
def test_SqlStatementSelect6() -> None:
sql = sql_utils.SqlStatementSelect("t")
sql.where = sql_utils.SqlConditionOr();
sql.where = sql_utils.SqlConditionOr()
sql.where.addCondition("cond1")
assert (str(sql) == "SELECT * FROM t WHERE (cond1)")