generated from Luis/nextjs-python-web-template
Added AP get repository #52
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)
|
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(
|
@router.get(
|
||||||
"/api/v1/entity_by_roles", response_model=List[Entity], tags=[Tags.entities]
|
"/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",
|
entity_did: str = "did:sov:test:120",
|
||||||
db: Session = Depends(sql_db.get_db),
|
db: Session = Depends(sql_db.get_db),
|
||||||
) -> Optional[sql_models.Entity]:
|
) -> 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
|
return entity
|
||||||
|
|
||||||
|
|
||||||
@@ -243,6 +231,8 @@ def get_rpc_by_role(db: Session, role: Role, path: str) -> Any:
|
|||||||
raise ClanError(f"No {role} found")
|
raise ClanError(f"No {role} found")
|
||||||
if len(matching_entities) > 1:
|
if len(matching_entities) > 1:
|
||||||
raise ClanError(f"More than one {role} found")
|
raise ClanError(f"More than one {role} found")
|
||||||
|
if len(matching_entities) == 0:
|
||||||
|
raise ClanError(f"No {role} found")
|
||||||
dlg = matching_entities[0]
|
dlg = matching_entities[0]
|
||||||
|
|
||||||
url = f"http://{dlg.ip}/{path}"
|
url = f"http://{dlg.ip}/{path}"
|
||||||
@@ -317,7 +307,6 @@ def get_all_eventmessages(
|
|||||||
# #
|
# #
|
||||||
##############################
|
##############################
|
||||||
@router.get("/emulate", response_class=HTMLResponse)
|
@router.get("/emulate", response_class=HTMLResponse)
|
||||||
@router.get("/emu", response_class=HTMLResponse)
|
|
||||||
def get_emulated_enpoints() -> HTMLResponse:
|
def get_emulated_enpoints() -> HTMLResponse:
|
||||||
from clan_cli.config import ap_url, c1_url, c2_url, dlg_url
|
from clan_cli.config import ap_url, c1_url, c2_url, dlg_url
|
||||||
|
|
||||||
|
|||||||
@@ -1,25 +1,15 @@
|
|||||||
import logging
|
import logging
|
||||||
from datetime import datetime
|
from datetime import datetime
|
||||||
from enum import Enum
|
|
||||||
from typing import List
|
from typing import List
|
||||||
|
|
||||||
from pydantic import BaseModel, Field, validator
|
from pydantic import BaseModel, Field, validator
|
||||||
|
|
||||||
|
from . import sql_models
|
||||||
|
from .db_types import Role, Status
|
||||||
|
|
||||||
log = logging.getLogger(__name__)
|
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):
|
class Machine(BaseModel):
|
||||||
name: str
|
name: str
|
||||||
status: Status
|
status: Status
|
||||||
@@ -72,8 +62,15 @@ class Entity(EntityBase):
|
|||||||
|
|
||||||
# define a custom getter function for roles
|
# define a custom getter function for roles
|
||||||
@validator("roles", pre=True)
|
@validator("roles", pre=True)
|
||||||
def get_roles(cls, v: List[EntityRoles]) -> List[Role]:
|
def get_roles(cls, v: List[sql_models.EntityRoles | Role]) -> List[Role]:
|
||||||
return [x.role for x in v]
|
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 import JSON, Boolean, Column, Enum, ForeignKey, Integer, String, Text
|
||||||
from sqlalchemy.orm import relationship
|
from sqlalchemy.orm import relationship
|
||||||
|
|
||||||
from .schemas import Role
|
from .db_types import Role
|
||||||
from .sql_db import Base
|
from .sql_db import Base
|
||||||
|
|
||||||
# Relationsship example
|
# Relationsship example
|
||||||
|
|||||||
Reference in New Issue
Block a user