blob: 367faf504ae6b2dba372b8fc1f4dcb805600344f (
plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
|
from typing import ClassVar
from dataclasses import Field
from serenitas.utils.db2 import dbconn
class Report:
_sql_insert: ClassVar[str]
_insert_queue: ClassVar[list]
_insert_fields: ClassVar[list]
_conn: ClassVar = dbconn("dawndb")
def __init_subclass__(cls, table_name: str):
cls._insert_fields = list(filter(cls.is_insert_field, cls.__annotations__))
place_holders = ",".join(["%s"] * len(cls._insert_fields))
cls._sql_insert = f"INSERT INTO {table_name}({','.join(cls._insert_fields)}) VALUES({place_holders}) ON CONFLICT DO NOTHING RETURNING *"
cls._insert_queue = []
@classmethod
def is_insert_field(cls, attr):
"""fields are insertable by default unless specified otherwise"""
match getattr(cls, attr, None):
case Field(metadata={"insert": False}) | Field(init=False):
return False
case _:
return True
@classmethod
def clear(cls):
cls._insert_queue.clear()
def stage(self):
self._insert_queue.append(
tuple([getattr(self, col) for col in self._insert_fields])
)
@classmethod
def commit(cls):
conn = cls._conn
with conn.cursor() as c:
c.executemany(cls._sql_insert, cls._insert_queue)
conn.commit()
cls._insert_queue.clear()
|
