Ukázka zpracování dat z SQLite databáze

Nejprve naimportujeme knihovny a potřebné moduly. Pro účely notebooku je třeba přidat cestu k modulu pywsdp do PATH.

import os
import sys
library_path = os.path.abspath(os.path.join("../../"))
sys.path.append(library_path)

from pywsdp.modules import CtiOS
from pywsdp.modules.CtiOS import OutputFormat

Příhlásíme se k modulu CtiOS. Pro tento účel využijeme testovací přístupové údaje do DP.

creds_test = ["WSTEST", "WSHESLO"]
ctios = CtiOS(creds_test, trial=True)

ctiOS        - INFO     - Logovaci zpravy ulozeny v adresari: /tmp/ctiOS

Definujeme cesty k vstupní databázi a k výstupnímu adresáři, kam budeme chtít updatovanou databázi uložit.

db_path = os.path.abspath(
    os.path.join(library_path, "tests", "data", "input", "ctios_template.db")
)

vystupni_adresar = os.path.abspath(
    os.path.join(library_path, "tests", "data", "output")
)

Pokud chceme změnit cestu k logovacímu adresáři, uděláme to následovně:

ctios.log_adresar = "/home/linduska/pywsdp/logs"

ctiOS        - INFO     - Logovaci adresar nastaven na cestu: /home/linduska/pywsdp/logs

Načteme identifikátory z databáze (pro účely ukázky omezíme počet načtených identifikátorů na 5).

parametry_ctiOS_db = ctios.nacti_identifikatory_z_db(db_path, "SELECT * FROM OPSUB order by ID LIMIT 5")

Pošleme požadavek a vypíšeme si statistiku zpracovaných identifikátorů.

slovnik, slovnik_chybnych = ctios.posli_pozadavek(parametry_ctiOS_db)

ctiOS        - INFO     -
ctiOS        - INFO     - Požadovaná akce byla úspěšně provedena.
ctiOS        - INFO     - POSIDENT /AOrBGOFR+4GtqsAvOXC/3kKd0OCEPBVRfEsIPHeW3CoGcU1dio8gmrulsH5/KEvySRvbRT2hlDOXICKukDLldMgNY5z+yASe8F1rqBlKug= USPESNE STAZEN
ctiOS        - INFO     - POSIDENT +lQ0Ne+Jhd5K4uvds1/KT8qqJcoA/eJ5OloIoqvCQi04X93kbmO62f/i2fqJgk62odrpl2ye04ue7tlKR93hckH4u46kVMoHQWmYF/rBWMA= USPESNE STAZEN
ctiOS        - INFO     - POSIDENT +VKfYaKhJDrRHUWeS2e7xO7g+HhFQapZMPa5kamaF/wSOs6r3xEPEVFhtUlWqcXO9GTopZEjqpsTbsKhZmN7VxYJSxvgcgTWMGdxYkQVCaU= USPESNE STAZEN
ctiOS        - INFO     - POSIDENT /ipU68hnjrq1J7ZDi4JBaAyBiKjxZzpGZSVIfv8yFMKSKy3PpT/lba/0c4YCdqvrexVCCAL0MxBbob3A75l2Tnkgw/HHXtm0+abSHVywLEM= USPESNE STAZEN
ctiOS        - INFO     - POSIDENT 00ep7V5fFBZoFMeOfcAOPMRtu0G9svLcsDyC/fYkJk3EbFIqbw2kpRaC1jjS1V8bjacEwrp/SdxXZXdtwDzFj3vKzXvHF7k7HiDpVtv6fPAUIS/cKLcyZ7viQd7pE5FE USPESNE STAZEN
ctiOS        - INFO     - Celkovy pocet dotazovanych identifikatoru na vstupu: 5
ctiOS        - INFO     - Pocet odstranenych duplicitnich identifikatoru: 0
ctiOS        - INFO     - Pocet pozadavku, do kterych byl dotaz rozdelen (pocet dotazu na server): 1
ctiOS        - INFO     - Realny uspesne zpracovanych identifikatoru: 5
ctiOS        - INFO     - Pocet neplatnych identifikatoru: 0
ctiOS        - INFO     - Pocet expirovanych identifikatoru: 0
ctiOS        - INFO     - Pocet identifikatoru k neexistujicim OS: 0

Uložíme získané osobní údaje do databáze.

vystup = ctios.uloz_vystup(
    slovnik, vystupni_adresar, OutputFormat.GdalDb
)

ctiOS        - INFO     - Vystup byl ulozen zde: /home/linduska/pywsdp/tests/data/output/ctios_21_20_37_10_01_2023.db

Podle vyplněného atributu jmeno (v případě fyzické osoby) nebo nazev (v případě právnické osoby nebo SJM) zkontrolujeme počet aktualizovaných řádků.

import sqlite3
con = sqlite3.connect(vystup)
cur = con.cursor()
for row in cur.execute("SELECT count(*) FROM OPSUB where jmeno is not null or nazev is not null"):
    assert row == (5,)
con.close()

Může se stát, že nechceme kopírovat celý soubor databáze do nového umístění, ale pouze chceme aktualizovat vstupní databázový soubor. Toho můžeme dosáhnout takto:

vystup = ctios.uloz_vystup_aktualizuj_db(
    slovnik
)