Skip to content
Snippets Groups Projects
Cube.py 2.88 KiB
Newer Older
  • Learn to ignore specific revisions
  • 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