Added AP get repository #52

Merged
merge-bot merged 3 commits from Qubasa-main into main 2024-01-14 15:01:37 +00:00
4 changed files with 29 additions and 30 deletions

View 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"

View File

@@ -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

View File

@@ -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
#########################

View File

@@ -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