jasima/majuna
3
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

lint: resolve mypy issues in app.cli.db

Browse source
This commit is contained in:
Iain Learmonth 2024-11-18 09:56:48 +00:00
parent c4e2ed364e
commit be111b9dde
1 changed files with 13 additions and 8 deletions
Download patch file Download diff file Expand all files Collapse all files

21
app/cli/db.py
View file

@ -4,7 +4,7 @@ import json
import logging import logging
import sys import sys
from collections import defaultdict from collections import defaultdict
from typing import List, Dict, Any from typing import Any, Callable, Dict, List, Type
from sqlalchemy import inspect from sqlalchemy import inspect
@ -20,8 +20,10 @@ from app.models.alarms import Alarm, AlarmState
from app.models.onions import Onion, Eotk from app.models.onions import Onion, Eotk
from app.models.tfstate import TerraformState from app.models.tfstate import TerraformState
Model = Type[db.Model] # type: ignore[name-defined]
# order matters due to foreign key constraints # order matters due to foreign key constraints
models = [ models: List[Model] = [
Group, Group,
Activity, Activity,
Pool, Pool,
@ -57,9 +59,12 @@ class ExportEncoder(json.JSONEncoder):
return super().default(o) return super().default(o)
def model_to_dict(model: db.Model) -> Dict[str, Any]: # type: ignore[name-defined] def model_to_dict(model: Model) -> Dict[str, Any]:
output = {} output = {}
for column in inspect(type(model)).columns: inspection = inspect(type(model))
if not inspection:
raise RuntimeError(f"Could not inspect model {model}")
for column in inspection.columns:
item = getattr(model, column.name) item = getattr(model, column.name)
output[f"{type(item).__name__}_{column.name}"] = item output[f"{type(item).__name__}_{column.name}"] = item
return output return output
@ -69,12 +74,12 @@ def db_export() -> None:
encoder = ExportEncoder() encoder = ExportEncoder()
output = defaultdict(list) output = defaultdict(list)
for model in models: for model in models:
for row in model.query.all(): # type: ignore[attr-defined] for row in model.query.all():
output[model.__name__].append(model_to_dict(row)) output[model.__name__].append(model_to_dict(row))
print(encoder.encode(output)) print(encoder.encode(output))
decoder = { decoder: Dict[str, Callable[[Any], Any]] = {
"AlarmState": lambda x: AlarmState.__getattribute__(AlarmState, x), "AlarmState": lambda x: AlarmState.__getattribute__(AlarmState, x),
"AutomationState": lambda x: AutomationState.__getattribute__(AutomationState, x), "AutomationState": lambda x: AutomationState.__getattribute__(AutomationState, x),
"bytes": lambda x: base64.decodebytes(x.encode('utf-8')), "bytes": lambda x: base64.decodebytes(x.encode('utf-8')),
@ -84,12 +89,12 @@ decoder = {
} }
def db_import_model(model: db.Model, data: List[Dict[str, Any]]) -> None: # type: ignore[name-defined] def db_import_model(model: Model, data: List[Dict[str, Any]]) -> None:
for row in data: for row in data:
new = model() new = model()
for col in row: for col in row:
type_name, col_name = col.split("_", 1) type_name, col_name = col.split("_", 1)
setattr(new, col_name, decoder.get(type_name, lambda x: x)(row[col])) # type: ignore[no-untyped-call] setattr(new, col_name, decoder.get(type_name, lambda x: x)(row[col]))
db.session.add(new) db.session.add(new)