events first design and class (4f562e14) · Commits · OCF / sdk

config/capif_sdk_config.json

+43 −0

Original line number	Diff line number	Diff line
		@@ -37,7 +37,50 @@
		"api-supported-features": "",
		"ue-ip-addr": "",
		"service-kpis": ""
		},
		"events":{
		"events": ["SERVICE_API_AVAILABLE"],
		"eventFilters": [
		{
		"apiIds": [
		"string"
		],
		"apiInvokerIds": [
		"string"
		],
		"aefIds": [
		"string"
		]
		}
		],
		"eventReq":{
		"immRep": true,
		"notifMethod": "string",
		"maxReportNbr": 5,
		"monDur": "DATE-time",
		"repPeriod": 5,
		"sampRatio": "integer from 1 to 100",
		"partitionCriteria": [
		"TAC"
		],
		"grpRepTime": 3,
		"notifFlag": "ACTIVATE",
		"notifFlagInstruct": {
		"bufferedNotifs":"SEND_ALL",
		"subscription":"CLOSE"
		},
		"mutingSetting": {
		"maxNoOfNotif":5,
		"durationBufferedNotif":5
		}
		},
		"requestTestNotification": true,
		"websockNotifConfig": {
		"websocketUri":"",
		"requestWebsocketUri": true
		}
		}

		},
		"provider": {
		"provider_folder": "",

opencapif_sdk/capif_event_feature.py

0 → 100644

+101 −0

Original line number	Diff line number	Diff line
		from capif_invoker_connector import capif_invoker_connector
		from capif_provider_connector import capif_provider_connector
		import os
		import logging
		import shutil
		from requests.auth import HTTPBasicAuth
		import urllib3
		from OpenSSL.SSL import FILETYPE_PEM
		from OpenSSL.crypto import (
		dump_certificate_request,
		dump_privatekey,
		PKey,
		TYPE_RSA,
		X509Req
		)
		import requests
		import json
		import warnings
		from requests.exceptions import RequestsDependencyWarning
		urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
		warnings.filterwarnings("ignore", category=RequestsDependencyWarning)
		# noqa: E501
		# Basic configuration of the logger functionality

		log_path = 'logs/sdk_logs.log'

		log_dir = os.path.dirname(log_path)

		if not os.path.exists(log_dir):
		os.makedirs(log_dir)

		logging.basicConfig(
		level=logging.NOTSET, # Minimum severity level to log
		# Log message format
		format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
		handlers=[
		logging.FileHandler(log_path), # Log to a file
		logging.StreamHandler() # Also display in the console
		]
		)

		class capif_invoker_event_feature(capif_invoker_connector):

		def create_subscription:

		invoker_capif_details = self.__load_invoker_api_details()

		subscriberId = invoker_capif_details["api_invoker_id"]

		path = self.capif_https_url + f"/{subscriberId}/subscriptions"

		payload = {
		"events": [
		"SERVICE_API_AVAILABLE",
		"string"
		],
		"eventFilters": [
		{
		"apiIds": [
		"string"
		],
		"apiInvokerIds": [
		"string"
		],
		"aefIds": [
		"string"
		]
		}
		],
		"eventReq": "string",
		"notificationDestination": f"{self.capif_callback_url}",
		"requestTestNotification": true,
		"websockNotifConfig": "string",
		}


		try:
		response = requests.post(
		url=path,
		json=payload,
		headers={"Content-Type": "application/json"},
		cert=cert,
		verify=os.path.join(self.invoker_folder, "ca.crt")
		)

		response.raise_for_status()

		return response.status_code, response.json()

		except Exception as e:
		self.logger.error("Unexpected error: %s", e)
		return None, {"error": f"Unexpected error: {e}"}


		def delete_subcription:

		def modify_subcription:

		def patch_subcription:


		No newline at end of file