generated from Luis/nextjs-python-web-template
Compare commits
4 Commits
8035fc15af
...
7045f945c8
| Author | SHA1 | Date | |
|---|---|---|---|
| 7045f945c8 | |||
| 44797a0bc3 | |||
| 52987625c8 | |||
| 8a3a593cc0 |
13
pkgs/clan-cli/clan_cli/webui/db_types.py
Normal file
13
pkgs/clan-cli/clan_cli/webui/db_types.py
Normal file
@@ -0,0 +1,13 @@
|
||||
from enum import Enum
|
||||
|
||||
|
||||
class Status(Enum):
|
||||
ONLINE = "online"
|
||||
OFFLINE = "offline"
|
||||
UNKNOWN = "unknown"
|
||||
|
||||
|
||||
class Role(Enum):
|
||||
PROSUMER = "service_prosumer"
|
||||
AP = "AP"
|
||||
DLG = "DLG"
|
||||
@@ -94,18 +94,6 @@ def create_entity(
|
||||
return sql_crud.create_entity(db, entity)
|
||||
|
||||
|
||||
@router.get(
|
||||
"/api/v1/entity_by_name_or_did",
|
||||
response_model=Optional[Entity],
|
||||
tags=[Tags.entities],
|
||||
)
|
||||
def get_entity_by_name_or_did(
|
||||
entity_name_or_did: str = "C1", db: Session = Depends(sql_db.get_db)
|
||||
) -> Optional[sql_models.Entity]:
|
||||
entity = sql_crud.get_entity_by_name_or_did(db, name=entity_name_or_did)
|
||||
return entity
|
||||
|
||||
|
||||
@router.get(
|
||||
"/api/v1/entity_by_roles", response_model=List[Entity], tags=[Tags.entities]
|
||||
)
|
||||
@@ -129,7 +117,7 @@ def get_entity_by_did(
|
||||
entity_did: str = "did:sov:test:120",
|
||||
db: Session = Depends(sql_db.get_db),
|
||||
) -> Optional[sql_models.Entity]:
|
||||
entity = sql_crud.get_entity_by_did(db, did=entity_did)
|
||||
entity = sql_crud.get_entity_by_name_or_did(db, name=entity_did)
|
||||
return entity
|
||||
|
||||
|
||||
@@ -243,6 +231,8 @@ def get_rpc_by_role(db: Session, role: Role, path: str) -> Any:
|
||||
raise ClanError(f"No {role} found")
|
||||
if len(matching_entities) > 1:
|
||||
raise ClanError(f"More than one {role} found")
|
||||
if len(matching_entities) == 0:
|
||||
raise ClanError(f"No {role} found")
|
||||
dlg = matching_entities[0]
|
||||
|
||||
url = f"http://{dlg.ip}/{path}"
|
||||
@@ -317,7 +307,6 @@ def get_all_eventmessages(
|
||||
# #
|
||||
##############################
|
||||
@router.get("/emulate", response_class=HTMLResponse)
|
||||
@router.get("/emu", response_class=HTMLResponse)
|
||||
def get_emulated_enpoints() -> HTMLResponse:
|
||||
from clan_cli.config import ap_url, c1_url, c2_url, dlg_url
|
||||
|
||||
|
||||
@@ -1,25 +1,15 @@
|
||||
import logging
|
||||
from datetime import datetime
|
||||
from enum import Enum
|
||||
from typing import List
|
||||
|
||||
from pydantic import BaseModel, Field, validator
|
||||
|
||||
from . import sql_models
|
||||
from .db_types import Role, Status
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class Status(Enum):
|
||||
ONLINE = "online"
|
||||
OFFLINE = "offline"
|
||||
UNKNOWN = "unknown"
|
||||
|
||||
|
||||
class Role(Enum):
|
||||
PROSUMER = "service_prosumer"
|
||||
AP = "AP"
|
||||
DLG = "DLG"
|
||||
|
||||
|
||||
class Machine(BaseModel):
|
||||
name: str
|
||||
status: Status
|
||||
@@ -72,8 +62,15 @@ class Entity(EntityBase):
|
||||
|
||||
# define a custom getter function for roles
|
||||
@validator("roles", pre=True)
|
||||
def get_roles(cls, v: List[EntityRoles]) -> List[Role]:
|
||||
return [x.role for x in v]
|
||||
def get_roles(cls, v: List[sql_models.EntityRoles | Role]) -> List[Role]:
|
||||
if (
|
||||
isinstance(v, list)
|
||||
and len(v) > 0
|
||||
and isinstance(v[0], sql_models.EntityRoles)
|
||||
):
|
||||
return [x.role for x in v] # type: ignore
|
||||
else:
|
||||
return v # type: ignore
|
||||
|
||||
|
||||
#########################
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
from sqlalchemy import JSON, Boolean, Column, Enum, ForeignKey, Integer, String, Text
|
||||
from sqlalchemy.orm import relationship
|
||||
|
||||
from .schemas import Role
|
||||
from .db_types import Role
|
||||
from .sql_db import Base
|
||||
|
||||
# Relationsship example
|
||||
|
||||
Reference in New Issue
Block a user