move traffic influence to core and update oai client (191891dc) · Commits · OOP / code / transformation-function-sdk

src/network/clients/oai/client.py

+86 −181

Original line number	Diff line number	Diff line
		@@ -8,38 +8,23 @@
		# - Giulio Carota (giulio.carota@eurecom.fr)
		##

		from typing import Dict

		import shortuuid
		from pydantic import ValidationError

		from src import logger
		from src.network.clients.oai.common import (
		OaiHttpError,
		OaiNetworkError,
		oai_as_session_with_qos_delete,
		oai_as_session_with_qos_get,
		oai_as_session_with_qos_post,
		oai_traffic_influence_delete,
		oai_traffic_influence_post,
		oai_traffic_influence_put,
		)
		from src.network.clients.oai.schemas import (
		CamaraQoDSessionInfo,
		CamaraTrafficInfluence,
		OaiAsSessionWithQosSubscription,
		)
		from src.network.clients.oai.utils import (
		as_session_with_qos_to_camara_qod,
		camara_qod_to_as_session_with_qos,
		camara_ti_to_3gpp_ti,
		)
		from src.network.core.network_interface import NetworkManagementInterface
		from src.network.core.schemas import (
		AsSessionWithQoSSubscription,
		CreateSession,
		CreateTrafficInfluence,
		FlowInfo,
		Snssai,
		TrafficInfluSub,
		)

		log = logger.get_logger(__name__)
		supportedQos = ["qos-e", "qos-s", "qos-m", "qos-l"]


		class OaiNefClient(NetworkManagementInterface):
		class NetworkManager(NetworkManagementInterface):
		def __init__(self, base_url: str, scs_as_id: str = None):
		"""
		Initialize Network Client for OAI Core Network
		@@ -50,185 +35,105 @@ class OaiNefClient(NetworkManagementInterface):
		try:
		super().__init__()
		self.base_url = base_url
		self.scs_as_id = shortuuid.uuid()
		self.scs_as_id = scs_as_id
		log.info(
		f"Initialized OaiNefClient with base_url: {self.base_url} and scs_as_id: {self.scs_as_id}"
		)

		except Exception as e:
		log.error(f"Failed to initialize OaiNefClient: {e}")
		raise e

		# implementation of the NetworkManagementInterface QoD Methods
		def create_qod_session(self, session_info: Dict) -> Dict:
		"""
		Creates a QoS session based on CAMARA QoD API input.
		It maps CAMARA QoD API POST /sessions to
		OAI NEF POST /3gpp-as-session-with-qos/v1/{scs_as_id}/subscriptions
		def core_specific_qod_validation(self, session_info: CreateSession):
		"""
		try:
		qod_input = CamaraQoDSessionInfo(**session_info)

		# convert CAMARA QoD to NEF AsSessionWithQos model and do POST
		nef_req = camara_qod_to_as_session_with_qos(qod_input)
		nef_res = oai_as_session_with_qos_post(
		self.base_url, self.scs_as_id, nef_req
		)
		Validates core-specific parameters for the session creation.

		# retrieve the NEF resource id
		if "self" in nef_res.keys():
		nef_url = nef_res["self"]
		nef_id = nef_url.split("subscriptions/")[1]
		else:
		raise OaiNetworkError(
		"No valid ID for the created resource was returned"
		)

		# create QoD session detail and return info with resource Id
		qod_input.sessionId = nef_id

		log.info(f"QoD session activated successfully [id={nef_id}]")

		return qod_input

		except ValidationError as e:
		raise OaiNetworkError("Could not validate QoD Session Info data") from e
		except KeyError as e:
		raise OaiNetworkError(f"Missing field in QoD Session Info data: {e}") from e
		except OaiHttpError as e:
		raise OaiNetworkError(
		f"The network could not enable the QoD Session. It returned {e}"
		) from e
		except OaiNetworkError as e:
		raise e
		args:
		session_info: The session information to validate.

		def get_qod_session(self, session_id: str) -> Dict:
		"""
		Retrieves details of a specific Quality on Demand (QoS) session.
		It maps CAMARA QoD API GET /sessions/{sessionId} to
		OAI NEF GET /3gpp-as-session-with-qos/v1/{scs_as_id}/subscriptions/{subscriptionId}
		raises:
		ValidationError: If the session information does not meet core-specific requirements.
		"""
		try:
		res = oai_as_session_with_qos_get(
		self.base_url, self.scs_as_id, session_id=session_id
		if session_info.qosProfile.root not in supportedQos:
		raise OaiValidationError(
		f"QoS profile {session_info.qosProfile} not supported by OAI, supported profiles are {supportedQos}"
		)
		nef_res = OaiAsSessionWithQosSubscription(**res)
		qod_info = as_session_with_qos_to_camara_qod(nef_res)

		log.info(f"QoD session retrived successfully [id={session_id}]")

		return qod_info
		except ValidationError as e:
		raise OaiNetworkError("Could not validate network response data") from e
		except OaiHttpError as e:
		raise OaiNetworkError(
		f"The network could not enable the QoD Session. It returned {e}"
		) from e
		except OaiNetworkError as e:
		raise e

		def delete_qod_session(self, session_id: str) -> None:
		"""
		Deletes a specific Quality on Demand (QoS) session.
		It maps CAMARA QoD API DELETE /sessions/{sessionId} to
		OAI NEF DELETE /3gpp-as-session-with-qos/v1/{scs_as_id}/subscriptions/{subscriptionId}
		if session_info.device is None or session_info.device.ipv4Address is None:
		raise OaiValidationError("OAI requires UE IPv4 Address to activate QoS")

		if session_info.applicationServer.ipv4Address is None:
		raise OaiValidationError("OAI requires App IPv4 Address to activate QoS")
		return

		def add_core_specific_qod_parameters(
		self,
		session_info: CreateSession,
		subscription: AsSessionWithQoSSubscription,
		) -> None:
		device_ip = _retrieve_ue_ipv4(session_info)
		server_ip = _retrieve_app_ipv4(session_info)

		# build flow descriptor in oai format using device ip and server ip
		flow_descriptor = f"permit out ip from {device_ip}/32 to {server_ip}/32"
		_add_qod_flow_descriptor(subscription, flow_descriptor)
		_add_qod_snssai(subscription, 1, "FFFFFF")
		subscription.dnn = "oai"

		def add_core_specific_ti_parameters(
		self,
		traffic_influence_info: CreateTrafficInfluence,
		subscription: TrafficInfluSub,
		):
		# todo oai add dnn, ssnai, afServiceId
		subscription.dnn = "oai"
		subscription.add_snssai(1, "FFFFFF")
		subscription.afServiceId = self.scs_as_id

		def core_specific_traffic_influence_validation(
		self, traffic_influence_info: CreateTrafficInfluence
		) -> None:
		"""
		try:
		oai_as_session_with_qos_delete(
		self.base_url, self.scs_as_id, session_id=session_id
		)
		Validates core-specific parameters for the session creation.

		log.info(f"QoD session deleted successfully [id={session_id}]")
		args:
		session_info: The session information to validate.

		except OaiHttpError as e:
		raise OaiNetworkError(
		f"The network could not enable the QoD Session. It returned {e}"
		) from e
		except OaiNetworkError as e:
		raise e

		# implementation of the NetworkManagementInterface Traffic Influence Methods
		def create_traffic_influence_resource(self, traffic_influence_info):
		try:
		ti_input = CamaraTrafficInfluence(**traffic_influence_info)

		# convert CAMARA TI to NEF TrafficInflSub model and do POST
		nef_req = camara_ti_to_3gpp_ti(ti_input)
		nef_res = oai_traffic_influence_post(self.base_url, self.scs_as_id, nef_req)

		# retrieve the NEF resource id
		if "self" in nef_res.keys():
		nef_url = nef_res["self"]
		nef_id = nef_url.split("subscriptions/")[1]
		else:
		raise OaiNetworkError(
		"No valid ID for the created resource was returned"
		raises:
		ValidationError: If the session information does not meet core-specific requirements.
		"""
		# Placeholder for core-specific validation logic
		# This method should be overridden by subclasses if needed

		if (
		traffic_influence_info.device is None
		or traffic_influence_info.device.ipv4Address is None
		):
		raise OaiValidationError(
		"OAI requires UE IPv4 Address to activate Traffic Influence"
		)

		# create TI session detail and return info with resource Id
		ti_input.trafficInfluenceID = nef_id

		log.info(f"Traffic Influence session activated successfully [id={nef_id}]")
		def _retrieve_ue_ipv4(session_info: CreateSession):
		return session_info.device.ipv4Address.root.privateAddress

		return ti_input

		except ValidationError as e:
		raise OaiNetworkError("Could not validate Traffic Influence data") from e
		except KeyError as e:
		raise OaiNetworkError(
		f"Missing field in Traffic Influence data: {e}"
		) from e
		except OaiHttpError as e:
		raise OaiNetworkError(
		f"The network could not enable the Traffic Influence Session. It returned {e}"
		) from e
		except OaiNetworkError as e:
		raise e
		def _retrieve_app_ipv4(session_info: CreateSession):
		return session_info.applicationServer.ipv4Address

		def delete_traffic_influence_resource(self, session_id):
		"""
		Deletes a specific Traffic Influence (TI) session.
		It maps CAMARA TI API DELETE /sessions/{sessionId} to
		OAI NEF DELETE /3gpp-traffic-influence/v1/{scs_as_id}/subscriptions/{subscriptionId}
		"""
		try:
		oai_traffic_influence_delete(
		self.base_url, self.scs_as_id, session_id=session_id

		def _add_qod_flow_descriptor(
		qos_sub: AsSessionWithQoSSubscription, flow_desriptor: str
		):
		qos_sub.flowInfo = list()
		qos_sub.flowInfo.append(
		FlowInfo(flowId=len(qos_sub.flowInfo) + 1, flowDescriptions=[flow_desriptor])
		)

		log.info(f"TI session deleted successfully [id={session_id}]")

		except OaiHttpError as e:
		raise OaiNetworkError(
		f"The network could not delete the TI session. It returned {e}"
		) from e
		except OaiNetworkError as e:
		raise e
		def _add_qod_snssai(qos_sub: AsSessionWithQoSSubscription, sst: int, sd: str = None):
		qos_sub.snssai = Snssai(sst=sst, sd=sd)

		def put_traffic_influence_resource(self, resource_id, traffic_influence_info):
		try:
		qod_input = CamaraTrafficInfluence(**traffic_influence_info)

		# convert CAMARA TI to NEF TrafficInflSub model and do POST
		nef_req = camara_ti_to_3gpp_ti(qod_input)
		updated_res = oai_traffic_influence_put(
		self.base_url, self.scs_as_id, resource_id, nef_req
		)

		log.info(
		f"Traffic Influence resource updated successfully [id={resource_id}]"
		)

		return updated_res

		except ValidationError as e:
		raise OaiNetworkError("Could not validate Traffic Influence data") from e
		except KeyError as e:
		raise OaiNetworkError(
		f"Missing field in Traffic Influence data: {e}"
		) from e
		except OaiHttpError as e:
		raise OaiNetworkError(
		f"The network could not update the Traffic Influence Session. It returned {e}"
		) from e
		except OaiNetworkError as e:
		raise e
		class OaiValidationError(Exception):
		pass

src/network/clients/oai/common.py

deleted100644 → 0

+0 −106

Original line number	Diff line number	Diff line
		##
		# Copyright (c) 2025 Netsoft Group, EURECOM.
		# All rights reserved.
		#
		# This file is part of the Open SDK
		#
		# Contributors:
		# - Giulio Carota (giulio.carota@eurecom.fr)
		##


		import requests
		from pydantic import BaseModel

		from src.network.clients.errors import NetworkPlatformError


		def _make_request(method: str, url: str, data=None):
		try:
		headers = None
		if method == "POST" or method == "PUT":
		headers = {
		"Content-Type": "application/json",
		"accept": "application/json",
		}
		elif method == "GET":
		headers = {
		"accept": "application/json",
		}
		response = requests.request(method, url, headers=headers, data=data)
		response.raise_for_status()
		if response.content:
		return response.json()
		except requests.exceptions.HTTPError as e:
		raise OaiHttpError(e) from e
		except requests.exceptions.ConnectionError as e:
		raise OaiHttpError("connection error") from e


		# QoD methods
		def oai_as_session_with_qos_post(
		base_url: str, scs_as_id: str, model_payload: BaseModel
		) -> dict:
		data = model_payload.model_dump_json(exclude_none=True)
		url = oai_as_session_with_qos_build_url(base_url, scs_as_id)
		return _make_request("POST", url, data=data)


		def oai_as_session_with_qos_get(base_url: str, scs_as_id: str, session_id: str) -> dict:
		url = oai_as_session_with_qos_build_url(base_url, scs_as_id, session_id)
		return _make_request("GET", url)


		def oai_as_session_with_qos_delete(base_url: str, scs_as_id: str, session_id: str):
		url = oai_as_session_with_qos_build_url(base_url, scs_as_id, session_id)
		return _make_request("DELETE", url)


		def oai_as_session_with_qos_build_url(
		base_url: str, scs_as_id: str, session_id: str = None
		):
		url = f"{base_url}/3gpp-as-session-with-qos/v1/{scs_as_id}/subscriptions"
		if session_id is not None and len(session_id) > 0:
		return f"{url}/{session_id}"
		else:
		return url


		## Traffic Influence methods
		def oai_traffic_influence_post(
		base_url: str, scs_as_id: str, model_payload: BaseModel
		) -> dict:
		data = model_payload.model_dump_json(exclude_none=True)
		url = oai_traffic_influence_build_url(base_url, scs_as_id)
		return _make_request("POST", url, data=data)


		def oai_traffic_influence_delete(base_url: str, scs_as_id: str, session_id: str):
		url = oai_traffic_influence_build_url(base_url, scs_as_id, session_id)
		return _make_request("DELETE", url)


		def oai_traffic_influence_put(
		base_url: str, scs_as_id: str, session_id: str, model_payload: BaseModel
		) -> dict:
		data = model_payload.model_dump_json(exclude_none=True)
		url = oai_traffic_influence_build_url(base_url, scs_as_id, session_id)
		return _make_request("PUT", url, data=data)


		def oai_traffic_influence_build_url(
		base_url: str, scs_as_id: str, session_id: str = None
		):
		url = f"{base_url}/3gpp-traffic-influence/v1/{scs_as_id}/subscriptions"
		if session_id is not None and len(session_id) > 0:
		return f"{url}/{session_id}"
		else:
		return url


		class OaiHttpError(Exception):
		pass


		class OaiNetworkError(NetworkPlatformError):
		pass

src/network/clients/oai/schemas.py

deleted100644 → 0

+0 −206

Original line number	Diff line number	Diff line
		##
		# Copyright (c) 2025 Netsoft Group, EURECOM.
		# All rights reserved.
		#
		# This file is part of the Open SDK
		#
		# Contributors:
		# - Giulio Carota (giulio.carota@eurecom.fr)
		##

		from typing import List, Optional

		from pydantic import BaseModel, Field


		class Snssai(BaseModel):
		sst: int = Field(default=1)
		sd: str = Field(default="FFFFFF")


		class TrafficFilter(BaseModel):
		flowId: int
		flowDescriptions: List[str]


		class OaiAsSessionWithQosSubscription(BaseModel):
		"""
		Represents the model to create an AsSessionWithQoS resource inside the OAI NEF.
		"""

		supportedFeatures: str = Field(default="12")
		dnn: str = Field(default="oai")
		snssai: Snssai
		flowInfo: List[TrafficFilter]
		ueIpv4Addr: str
		notificationDestination: str
		qosReference: str
		self: Optional[str] = None
		qosDuration: Optional[int] = None

		def add_flow_descriptor(self, flow_desriptor: str):
		self.flowInfo = list()
		self.flowInfo.append(
		TrafficFilter(
		flowId=len(self.flowInfo) + 1, flowDescriptions=[flow_desriptor]
		)
		)

		def add_snssai(self, sst: int, sd: str = None):
		self.snssai = Snssai(sst=sst, sd=sd)


		class PortRange(BaseModel):
		from_: int = Field(alias="from")
		to: int

		class Config:
		populate_by_name = True


		class Ports(BaseModel):
		ranges: Optional[List[PortRange]] = None
		ports: Optional[List[int]] = None


		class Ipv4Address(BaseModel):
		publicAddress: str
		publicPort: Optional[int] = None


		class Device(BaseModel):
		phoneNumber: Optional[str] = None
		networkAccessIdentifier: Optional[str] = None
		ipv4Address: Optional[Ipv4Address] = None
		ipv6Address: Optional[str] = None


		class ApplicationServer(BaseModel):
		ipv4Address: Optional[str] = None
		ipv6Address: Optional[str] = None


		class SinkCredential(BaseModel):
		credentialType: Optional[str] = None


		class CamaraQoDSessionInfo(BaseModel):
		"""
		Represents the input data for creating a QoD session.
		"""

		duration: int
		qosProfile: str
		applicationServer: ApplicationServer

		device: Optional[Device] = None
		devicePorts: Optional[Ports] = None
		applicationServerPorts: Optional[Ports] = None
		sink: Optional[str] = None
		sinkCredential: Optional[SinkCredential] = None

		# fields only applicable to sessionInfo in responses:
		sessionId: Optional[str] = None
		startedAt: Optional[int] = None
		expiresAt: Optional[int] = None
		qosStatus: Optional[str] = None
		statusInfo: Optional[str] = None

		class Config:
		populate_by_name = True

		def retrieve_ue_ipv4(self):
		if self.device is not None and self.device.ipv4Address is not None:
		return self.device.ipv4Address.publicAddress
		else:
		raise KeyError("device.ipv4Address.publicAddress")

		def retrieve_app_ipv4(self):
		if self.applicationServer.ipv4Address is not None:
		return self.applicationServer.ipv4Address
		else:
		raise KeyError("applicationServer.ipv4Address")

		def add_server_ipv4(self, ipv4: str):
		self.applicationServer = ApplicationServer(ipv4Address=ipv4)

		def add_ue_ipv4(self, ipv4: str):
		if self.device is None:
		self.device = Device()
		if self.device.ipv4Address is None:
		self.device.ipv4Address = Ipv4Address(publicAddress=ipv4)


		## traffic_influence schemas


		class SourceTrafficFilters(BaseModel):
		sourcePort: int


		class DestinationTrafficFilters(BaseModel):
		destinationPort: int
		destinationProtocol: str


		class TrafficRoute(BaseModel):
		dnai: str


		class NotificationSink(BaseModel):
		sink: Optional[str] = None
		sinkCredential: Optional[SinkCredential] = None


		class TrafficInfluSub(BaseModel): # Replace with a meaningful name
		afServiceId: str
		afAppId: str
		dnn: str
		snssai: Snssai
		trafficFilters: List[TrafficFilter]
		ipv4Addr: str
		notificationDestination: str
		trafficRoutes: List[TrafficRoute]
		suppFeat: str

		def add_flow_descriptor(self, flow_desriptor: str):
		self.trafficFilters = list()
		self.trafficFilters.append(
		TrafficFilter(
		flowId=len(self.trafficFilters) + 1, flowDescriptions=[flow_desriptor]
		)
		)

		def add_traffic_route(self, dnai: str):
		self.trafficRoutes = list()
		self.trafficRoutes.append(TrafficRoute(dnai=dnai))

		def add_snssai(self, sst: int, sd: str = None):
		self.snssai = Snssai(sst=sst, sd=sd)


		class CamaraTrafficInfluence(BaseModel):
		trafficInfluenceID: Optional[str] = None
		apiConsumerId: str
		appId: str
		appInstanceId: str
		edgeCloudRegion: Optional[str] = None
		edgeCloudZoneId: str
		sourceTrafficFilters: Optional[SourceTrafficFilters] = None
		destinationTrafficFilters: Optional[DestinationTrafficFilters] = None
		notificationUri: Optional[str] = None
		notificationAuthToken: Optional[str] = None
		device: Device
		notificationSink: Optional[NotificationSink] = None

		def retrieve_ue_ipv4(self):
		if self.device is not None and self.device.ipv4Address is not None:
		return self.device.ipv4Address.publicAddress
		else:
		raise KeyError("device.ipv4Address.publicAddress")

		def add_ue_ipv4(self, ipv4: str):
		if self.device is None:
		self.device = Device()
		if self.device.ipv4Address is None:
		self.device.ipv4Address = Ipv4Address(publicAddress=ipv4)

src/network/clients/oai/utils.py

deleted100644 → 0

+0 −88

Original line number	Diff line number	Diff line
		##
		# Copyright (c) 2025 Netsoft Group, EURECOM.
		# All rights reserved.
		#
		# This file is part of the Open SDK
		#
		# Contributors:
		# - Giulio Carota (giulio.carota@eurecom.fr)
		##


		from src.network.clients.oai.schemas import (
		CamaraQoDSessionInfo,
		CamaraTrafficInfluence,
		OaiAsSessionWithQosSubscription,
		TrafficInfluSub,
		)


		def camara_qod_to_as_session_with_qos(
		qod_input: CamaraQoDSessionInfo,
		) -> OaiAsSessionWithQosSubscription:
		device_ip = qod_input.retrieve_ue_ipv4()
		server_ip = qod_input.retrieve_app_ipv4()

		# Extract callback sink and QoS profile
		sink_url = qod_input.sink
		qos_profile = qod_input.qosProfile

		# build flow descriptor in oai format using device ip and server ip
		flow_descriptor = f"permit out ip from {device_ip}/32 to {server_ip}/32"

		# create the nef request model
		nef_req = OaiAsSessionWithQosSubscription.construct()
		nef_req.ueIpv4Addr = device_ip
		nef_req.notificationDestination = sink_url
		nef_req.add_flow_descriptor(flow_desriptor=flow_descriptor)
		nef_req.qosReference = qos_profile
		nef_req.add_snssai(1, "FFFFFF")

		# the qos duration feature is not available yet in oai
		# nef_req.qosDuration = qod_input.duration

		return nef_req


		def as_session_with_qos_to_camara_qod(
		nef_input: OaiAsSessionWithQosSubscription,
		) -> CamaraQoDSessionInfo:
		# create the camara qod model

		qod_info = CamaraQoDSessionInfo.construct()

		flowDesc = nef_input.flowInfo[0].flowDescriptions[0]
		serverIp = flowDesc.split("to ")[1].split("/32")[0]

		qod_info.add_server_ipv4(serverIp)
		qod_info.qosProfile = nef_input.qosReference
		qod_info.add_ue_ipv4(nef_input.ueIpv4Addr)
		qod_info.sink = nef_input.notificationDestination
		qod_info.duration = nef_input.qosDuration

		return qod_info


		def camara_ti_to_3gpp_ti(ti_input: CamaraTrafficInfluence) -> TrafficInfluSub:

		device_ip = ti_input.retrieve_ue_ipv4()
		server_ip = (
		ti_input.appInstanceId
		) # assume that the instance id corresponds to its IPv4 address
		sink_url = ti_input.notificationSink.sink
		edge_zone = ti_input.edgeCloudZoneId

		# build flow descriptor in oai format using device ip and server ip
		flow_descriptor = f"permit out ip from {device_ip}/32 to {server_ip}/32"

		nef_traffic_influence = TrafficInfluSub.model_construct()
		nef_traffic_influence.afAppId = ti_input.appId
		nef_traffic_influence.afServiceId = "aa"
		nef_traffic_influence.ipv4Addr = device_ip
		nef_traffic_influence.notificationDestination = sink_url
		nef_traffic_influence.add_flow_descriptor(flow_desriptor=flow_descriptor)
		nef_traffic_influence.add_traffic_route(dnai=edge_zone)
		nef_traffic_influence.add_snssai(1, "FFFFFF")
		nef_traffic_influence.dnn = "oai"

		return nef_traffic_influence

src/network/core/common.py

+31 −1

Original line number	Diff line number	Diff line
		# -- coding: utf-8 --
		# Common utilities (errors, HTTP helpers) used by the Open5GS client implementation (client.py).
		# Common utilities (errors, HTTP helpers) used by the core network interface (network_interface.py).

		import requests
		from pydantic import BaseModel
		@@ -60,5 +60,35 @@ def as_session_with_qos_build_url(
		return url


		# Traffic Influence Methods
		def traffic_influence_post(
		base_url: str, scs_as_id: str, model_payload: BaseModel
		) -> dict:
		data = model_payload.model_dump_json(exclude_none=True)
		url = traffic_influence_build_url(base_url, scs_as_id)
		return _make_request("POST", url, data=data)


		def traffic_influence_delete(base_url: str, scs_as_id: str, session_id: str):
		url = traffic_influence_build_url(base_url, scs_as_id, session_id)
		return _make_request("DELETE", url)


		def traffic_influence_put(
		base_url: str, scs_as_id: str, session_id: str, model_payload: BaseModel
		) -> dict:
		data = model_payload.model_dump_json(exclude_none=True)
		url = traffic_influence_build_url(base_url, scs_as_id, session_id)
		return _make_request("PUT", url, data=data)


		def traffic_influence_build_url(base_url: str, scs_as_id: str, session_id: str = None):
		url = f"{base_url}/3gpp-traffic-influence/v1/{scs_as_id}/subscriptions"
		if session_id is not None and len(session_id) > 0:
		return f"{url}/{session_id}"
		else:
		return url


		class CoreHttpError(Exception):
		pass

Admin message