ADD endpoint to obtain information about insert types

api/patch_request.py

@@ -8,7 +8,7 @@ from starlette import status
 
 from api import db, get_conn
 from model.patch_request import APIPatchRequestAction, PatchRequestFormRequest, InformalPatchRequestFormRequest, \
-    PatchRequestAction, PatchActionType
+    PatchRequestAction, PatchActionType, InsertType, InsertPort
 from model.settings import settings
 from model.wapi.cntl import APIToken, Mgr
 from model.wapi.nd import Port, Module, Room, Bldg, Site
@@ -36,6 +36,99 @@ async def get_available_sites(token: APIToken = Depends(check_auth), conn=Depend
     return [r[site][0] for site in available_fq_names]
 
 
+@router.get('/insert_types')
+async def get_insert_types(
+        #token: APIToken = Depends(check_auth),
+        conn=Depends(get_conn)) -> list[InsertType]:
+    #user = Mgr.check_token(conn, token)
+
+    searched_types = [
+        'Insert-284-1 ETH-10MB',
+        'Insert-3635-1 ISDN',
+        'Insert-1711 4*ISDN',
+        'Insert-6554-1 2*ISDN',
+        'Insert-4584-1 1*ISDN',
+        'Insert-3636-1 ETH-100MB',
+        'Insert-6553-1 ETH-100MB',
+        'Insert-3641-1 ISDN/ETH-100MB',
+        'Insert-1100-1 2*ISDN/ETH-100MB',
+        'Insert-73-1 ETH-10MB',
+        'Insert-47-1 ETH-100MB',
+        'Insert-6548-1 ETH-1GB',
+        'Insert-4584-1 ETH-1GB',
+        'Insert-1188-1 ETH-10GB',
+    ]
+
+    ta = [{
+            'name': 'ndcfg.module_type.list',
+            'idx': 'module_type_list',
+            'old': {
+                'name_list': searched_types
+            }
+        },
+        {
+            'name': 'ndcfg.p_port2module_type.list',
+            'idx': 'pp2mt_list',
+            'inner_join_ref': {
+                'module_type_list': 'default'
+            },
+        }
+    ]
+
+    result = execute_wapi_function(conn, ta, dry_mode=True, superuser=True)
+
+    module_types: dict[str, InsertType] = {}
+
+    for api_type in result['module_type_list']:
+        module_types[api_type['name']] = InsertType(name=api_type['name'])
+
+    for api_port in result['pp2mt_list']:
+        if api_port['port_type'].lower().startswith('festkabel'):
+            continue
+        tmp_ports = [api_port]
+
+        cmp_ports: list[dict] = []
+        for tmp_port in tmp_ports:
+            if not tmp_port['port_name_suffix'].startswith('{'):
+                cmp_ports.append(tmp_port)
+                continue
+            suffix_list = list(tmp_port['port_name_suffix'].replace('{', '').replace('}', ''))
+            tmp_port['port_count'] -= (len(suffix_list) - 1) if len(suffix_list) > 0 else 0
+            for i in suffix_list if len(suffix_list) > 0 else ['']:
+                cmp_ports.append({
+                    **tmp_port,
+                    'port_name_suffix': i
+                })
+        tmp_ports = cmp_ports
+
+        cmp_ports: list[dict] = []
+        for tmp_port in tmp_ports:
+            if not tmp_port['port_name_prefix'].startswith('{'):
+                cmp_ports.append(tmp_port)
+                continue
+            prefix_list = list(tmp_port['port_name_prefix'].replace('{', '').replace('}', ''))
+            tmp_port['port_count'] -= (len(prefix_list) - 1) if len(prefix_list) > 0 else 0
+            for i in prefix_list if len(prefix_list) > 0 else ['']:
+                cmp_ports.append({
+                    **tmp_port,
+                    'port_name_prefix': i
+                })
+        tmp_ports = cmp_ports
+
+        for p in tmp_ports:
+            module_types[p['module_type']].ports.append(
+                InsertPort(type=p['port_type'],
+                           type_group=p['port_type_group'],
+                           count=p['port_count'],
+                           proto=p['port_proto'],
+                           name_prefix=p['port_name_prefix'],
+                           name_suffix=p['port_name_suffix']
+                           )
+            )
+
+    return list(module_types.values())
+
+
 @router.post('/informal')
 async def handle_informal_request(informal_patch_request_form_request: InformalPatchRequestFormRequest,
                                   token: APIToken = Depends(check_auth),

model/patch_request.py

 from typing import Optional
 
-from pydantic import BaseModel, Field
+from pydantic import BaseModel, Field, computed_field
 from enum import Enum
 
 from model.wapi.nd import BCD, Bldg, Module, Room, Port, Site
@@ -46,3 +46,25 @@ class InformalPatchRequestFormRequest(BaseModel):
     site: Site = Field(default=None)
     comment: str = Field()
     reply_to: Optional[str] = Field(default=None)
+
+
+class InsertPort(BaseModel):
+    type: str = Field()
+    type_group: str = Field()
+    count: int = Field(default=1)
+    proto: str = Field()
+    name_prefix: str = Field(default='')
+    name_suffix: str = Field(default='')
+
+    @computed_field
+    def name(self) -> str:
+        return f'{self.name_prefix}%{{module_name}}{self.name_suffix}'
+
+    @computed_field
+    def can_assign_bcd(self) -> bool:
+        return self.proto not in ['ISDN']
+
+
+class InsertType(BaseModel):
+    name: str = Field()
+    ports: list[InsertPort] = Field(default=[])