Začetek ORM in spremembe modela na predavanjih

Author: jaanos

predavanja/primeri/banka/model.py

@@ -1,32 +1,52 @@
+from auth import auth
 from dataclasses import dataclass
 from datetime import datetime
-
+from orm import povezi, stolpec, Entiteta, Funkcija
 
 @dataclass
-class Kraj:
-    posta: int
-    kraj: str
+class Kraj(Entiteta):
+    posta: int = stolpec(glavni_kljuc=True)
+    kraj: str = stolpec(obvezen=True)
 
 
 @dataclass
-class Oseba:
-    emso: str
-    ime: str
-    priimek: str
-    naslov: str
-    kraj: Kraj
+class Oseba(Entiteta):
+    emso: str = stolpec(glavni_kljuc=True)
+    ime: str = stolpec(obvezen=True)
+    priimek: str = stolpec(obvezen=True)
+    naslov: str = stolpec(obvezen=True)
+    kraj: Kraj = stolpec(obvezen=True)
 
 
 @dataclass
-class Racun:
-    stevilka: int
-    lastnik: Oseba
+class Racun(Entiteta):
+    stevilka: int = stolpec(glavni_kljuc=True, stevec=True)
+    lastnik: Oseba = stolpec(obvezen=True)
 
 
 @dataclass
-class Transakcija:
-    id: int
-    racun: Racun
-    znesek: int
-    cas: datetime = None
-    opis: str = None
+class Transakcija(Entiteta):
+    id: int = stolpec(glavni_kljuc=True, stevec=True)
+    racun: Racun = stolpec(obvezen=True)
+    znesek: int = stolpec(obvezen=True)
+    cas: datetime = stolpec(privzeto=Funkcija("NOW()"), obvezen=True)
+    opis: str = stolpec()
+
+
+def ustvari_tabele(izbrisi=False, ce_ne_obstajajo=False):
+    with conn.transaction():
+        if izbrisi:
+            izbrisi_tabele(ce_ne_obstajajo)
+        for entiteta in ENTITETE:
+            entiteta.ustvari_tabelo(ce_ne_obstaja=ce_ne_obstajajo)
+
+
+def izbrisi_tabele(ce_obstajajo=False):
+    with conn.transaction():
+        for entiteta in reversed(ENTITETE):
+            entiteta.izbrisi_tabelo(ce_obstajajo)
+
+
+ENTITETE = (Kraj, Oseba, Racun, Transakcija)
+conn = povezi(**auth)
+ustvari_tabele(ce_ne_obstajajo=True)

predavanja/primeri/banka/orm.py

@@ -0,0 +1,110 @@
+from dataclasses import field, fields
+from datetime import datetime
+from psycopg import connect, sql, errors
+import csv
+
+
+TIPI = {
+    int: "INTEGER",
+    str: "TEXT",
+    datetime: "TIMESTAMP(0)"
+}
+
+
+LASTNOSTI = {
+    "glavni_kljuc": "PRIMARY KEY",
+    "obvezen": "NOT NULL",
+    "enolicen": "UNIQUE"
+}
+
+
+def povezi(*largs, **kwargs):
+    """
+    Vzpostavi povezavo z bazo.
+    """
+    global conn
+    conn = connect(*largs, **kwargs, autocommit=True)
+    return conn
+
+
+def stolpec(privzeto=None, glavni_kljuc=False, stevec=False, obvezen=False,
+            enolicen=False):
+    return field(default=privzeto,
+                 metadata=dict(glavni_kljuc=glavni_kljuc, stevec=stevec,
+                               obvezen=obvezen, enolicen=enolicen))
+
+
+def tip(stolpec):
+    t = stolpec.type
+    while issubclass(t, Entiteta):
+        t = t.glavni_kljuc().type
+    return TIPI[t]
+
+
+class Funkcija:
+    def __init__(self, niz):
+        self.niz = sql.SQL(niz)
+
+
+class Entiteta:
+    @classmethod
+    def ustvari_tabelo(cls, pobrisi=False, ce_ne_obstaja=False):
+        with conn.transaction():
+            if pobrisi:
+                cls.izbrisi_tabelo(ce_ne_obstaja)
+            with conn.cursor() as cur:
+                niz = sql.SQL("""
+                    CREATE TABLE {ce_ne_obstaja} {tabela} ({stolpci});
+                """).format(
+                    ce_ne_obstaja=sql.SQL("IF NOT EXISTS" if ce_ne_obstaja else ""),
+                    tabela=sql.Identifier(cls.tabela()),
+                    stolpci=sql.SQL(", ").join(
+                        sql.SQL("{ime} {tip} {lastnosti} {privzeto} {referenca}").format(
+                            ime=sql.Identifier(stolpec.name),
+                            tip=sql.SQL("SERIAL" if stolpec.metadata["stevec"] else tip(stolpec)),
+                            lastnosti=sql.SQL(" ").join(
+                                sql.SQL(niz) for lastnost, niz in LASTNOSTI.items()
+                                if stolpec.metadata[lastnost]
+                            ),
+                            privzeto=sql.SQL("DEFAULT {vrednost}").format(
+                                vrednost=stolpec.default.niz if isinstance(stolpec.default, Funkcija)
+                                    else sql.Literal(stolpec.default)
+                            ) if stolpec.default else sql.SQL(""),
+                            referenca=sql.SQL("REFERENCES {tab}({st})").format(
+                                tab=sql.Identifier(stolpec.type.tabela()),
+                                st=sql.Identifier(stolpec.type.glavni_kljuc().name)
+                            ) if issubclass(stolpec.type, Entiteta) else sql.SQL("")
+                        )
+                        for stolpec in fields(cls)
+                    ))
+                cur.execute(niz)
+
+    @classmethod
+    def izbrisi_tabelo(cls, ce_obstaja=False):
+        with conn.transaction():
+            with conn.cursor() as cur:
+                cur.execute(sql.SQL("DROP TABLE {ce_obstaja} {tabela};").format(
+                    ce_obstaja=sql.SQL("IF EXISTS" if ce_obstaja else ""),
+                    tabela=sql.Identifier(cls.tabela())
+                ))
+
+    @classmethod
+    def tabela(cls):
+        if not hasattr(cls, "TABELA"):
+            cls.TABELA = cls.__name__.lower()
+        return cls.TABELA
+
+    @classmethod
+    def glavni_kljuc(cls):
+        if not hasattr(cls, "GLAVNI_KLJUC"):
+            cls.GLAVNI_KLJUC = next(stolpec for stolpec in fields(cls) if stolpec.metadata["glavni_kljuc"])
+        return cls.GLAVNI_KLJUC
+
+    @classmethod
+    def uvozi_podatke(cls):
+        with open(f"podatki/{cls.tabela()}.csv"):
+            rd = csv.reader(f)
+            stolpci = next(rd)
+            for vrstica in rd:
+                # vstavi podatke v bazo
+                pass