126 lines
4.3 KiB
Python
126 lines
4.3 KiB
Python
"""! @brief Defines the low-level DAO class to handle addresses."""
|
||
##
|
||
# @file AddressDAO.py
|
||
#
|
||
# @brief Defines the AddressDAO class.
|
||
#
|
||
# @section description_addressdao Description
|
||
# Defines the low-lever DAO class to handle CRUD operations on addresses.
|
||
# - AddressDAO (DAO class)
|
||
#
|
||
# @section libraries_addressdao Libraries/Modules
|
||
# - <a href="https://pypi.org/project/mariadb/">mariadb</a> Python module
|
||
# - MariaDB Connector/Python enables python programs to access MariaDB and MySQL databases, using an API which is compliant with the Python DB API 2.0 (PEP-249).
|
||
# - <a href="https://docs.python.org/3/library/json.html">json</a> Python standard library
|
||
# - JSON encoder and decoder.
|
||
# - <a href="https://docs.python.org/3/library/random.html">DbManager</a> Local class
|
||
# - Singleton class to provide for the database connection.
|
||
#
|
||
# @section notes_addressdao Notes
|
||
# - None.
|
||
#
|
||
# @section todo_addressdao TODO
|
||
# - None.
|
||
#
|
||
# @section author_addressdao Author(s)
|
||
# - Created by Linuxero on 03/14/2025.
|
||
# - Modified by Linuxero on 03/14/2025.
|
||
#
|
||
# Copyright (c) 2025 Schnaxero. All rights reserved.
|
||
|
||
from .DbManager import DbManager
|
||
import mariadb
|
||
import json
|
||
|
||
|
||
class AddressDAO:
|
||
"""! The AddressDAO class.
|
||
Defines a low-level class utilized for DAO.
|
||
Handles the address CRUD operations.
|
||
"""
|
||
__cur = None
|
||
def __init__(self):
|
||
"""! The AddressDAO class initializer.
|
||
"""
|
||
#print(f"*** File: {__file__}, init()")
|
||
self.__con = DbManager().getConnection()
|
||
|
||
if self.__con:
|
||
self.__cur = self.__con.cursor()
|
||
|
||
|
||
|
||
|
||
def __importPlz(self):
|
||
"""! Utility function to import zipcodes from an external databases.
|
||
"""
|
||
with open("pfad zur datei", "r") as plz:
|
||
postcodes = json.load(plz)
|
||
irgendwas = ""
|
||
try:
|
||
for i in postcodes:
|
||
test =i["plz_name"].split(",")
|
||
for town in test:
|
||
if "u.a" in town:
|
||
town = town[:-4]
|
||
town = town.strip()
|
||
if town:
|
||
print(f"PROCESSING {i['name']} {town}")
|
||
self.__cur.callproc("addZipCodes", (i["name"], town, irgendwas,))
|
||
#self.__cur.callproc("addZipCodes", ("56271", "Kleinmaischeid", irgendwas,))
|
||
except mariadb.OperationalError as e:
|
||
print(f"Database Error: {e}")
|
||
finally:
|
||
self.__con.commit()
|
||
print("FINISHED")#
|
||
|
||
def __importCountry(self):
|
||
"""! Utility function to import countries information from an external databases.
|
||
"""
|
||
with open("pfad zur datei", "r") as country:
|
||
countries = json.load(country)
|
||
old = ""
|
||
try:
|
||
for i in countries["daten"]:
|
||
if i[4] == "–":
|
||
continue
|
||
elif i[8] == "YU":
|
||
continue
|
||
elif i[4] == old:
|
||
continue
|
||
elif i[4] == "Serbien und Montenegro":
|
||
continue
|
||
elif i[4] == "Bosnien und Herzegowina":
|
||
continue
|
||
|
||
print(i[4], i[3], i[2], i[8], i[7])
|
||
|
||
self.__cur.execute("INSERT INTO country (country, countryshort, nationality, iso2, iso3) VALUES (%s, %s, %s, %s, %s)", (i[4], i[3], i[2], i[8], i[7]))
|
||
old = i[4]
|
||
except mariadb.OperationalError as e:
|
||
print(f"Database Error: {e}")
|
||
finally:
|
||
self.__con.commit()
|
||
print("FINISHED")#
|
||
|
||
|
||
|
||
|
||
def getAddressData(self, all = True, zipcode = None):
|
||
"""! Loads available addresses in the program.
|
||
@param all Filter to get specific addresses as a boolean.
|
||
@param zipcode Specific zipcode pattern.
|
||
@return Found addresses as a dictionary on success, None on failure.
|
||
"""
|
||
try:
|
||
if self.__cur:
|
||
self.__cur.callproc("getAddress", (all, zipcode,))
|
||
self.__data = self.__cur.fetchall()
|
||
return self.__data
|
||
else:
|
||
return None
|
||
except mariadb.Error as e:
|
||
print(str(e))
|
||
|
||
|