Newer
Older
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
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
import logging as log
from typing import List
from datetime import datetime, timezone
from zoneinfo import ZoneInfo
from pydantic import BaseModel
from sqlalchemy.orm import Session
from pilab.crud.Util import from_json, update_attrs
from pilab.schemas.Cube import CubesType
from pilab.schemas.Pi import PiType
from pilab.crud.Hardware import Controller, Switch
from pilab.crud.Pi import Pi
from pilab.events import meta
log.getLogger('sqlalchemy.engine').setLevel(log.WARNING)
class Cube(object):
@staticmethod#DONE
def get(db: Session, cube_id: int = None, switch_id: int = None, controller_id : int = None):
if cube_id:
db_cube = db.query(CubesType).filter(CubesType.id == cube_id).first()
elif switch_id:
db_cube = db.query(CubesType).filter(CubesType.switch_id == switch_id).first()
elif controller_id:
db_cube = db.query(CubesType).filter(CubesType.controller_id == controller_id).first()
controller = Controller.get(db, host_id = db_cube.controller_id) if db_cube.controller_id else None
switch = Switch.get(db, host_id = db_cube.switch_id)
pis = Pi.getAll(db, cube_id = db_cube.id)
return meta.Cube(
id=db_cube.id,
controller=controller,
switch=switch,
pis=pis,
)
@staticmethod#DONE
def create(db: Session, cube: meta.Cube):
if cube.controller:
controller_id = Controller.create(db, cube.controller).id
else:
controller_id = None
switch_id = Switch.create(db, cube.switch).id
db_cube = CubesType(
id=cube.id,
controller_id=controller_id,
switch_id=switch_id,
)
db.add(db_cube)
db.flush()
for pi in cube.pis:
Pi.create(db, pi)
return db_cube.id
@staticmethod#DONE
def delete(db: Session, cube_id: int):
db_cube = db.query(CubesType).filter(CubesType.id == cube_id).first()
db_pis = db.query(PiType).filter(
PiType.cube_id == cube_id).all()
for db_pi in db_pis:
Pi.delete(db, db_pi.host_id)
db.delete(db_cube)
if db_cube.controller_id:
Controller.delete(db, db_cube.controller_id)
Switch.delete(db, db_cube.switch_id)
db.flush()
@staticmethod
def get_ids(db: Session):
ids = []
db_ids = db.query(CubesType.id).all()
for i, in db_ids:
ids.append(i)
return ids
@staticmethod
def is_valid_id(db: Session, cube_id: int):
cube = db.query(CubesType).filter(
CubesType.id == cube_id).first()
return True if cube else False
@staticmethod
def get_all(db: Session):
ids = Cube.get_ids(db)
cubes = []
for i in ids:
cube = Cube.get(db, i)
cubes.append(cube)
return cubes