""" AUTHOR: Khushal P Soonderji DATE: Monday, 10th Feb., 2025. OBJECTIVE: To handle all MikroTik configuration from one place. REFERENCES: N/A DOWNLOADS: N/A """ # ***************************************************************************************************************** # ***** **** # *** IMPORT *** # ***** **** # ***************************************************************************************************************** # To make sibling directories accessible for imports: import sys sys.path.append(".") sys.path.append("..") # System-level activities: import distro import socket import platform # My async utils: from utils_v2.string import json from utils_v2.date_time import date_time from utils_v2.database.async_mongo_v2 import AsyncMongo from utils_v2.api.async_quart import describe_exception # To make very controlled API calls: from utils_v2.rest.controllers.async_base import AsyncREST from utils_v2.rest.controllers.auth import BasicAuth from utils_v2.rest.models.api_call import ApiResponse # To work with IP Addresses: import ipaddress # To work with datatypes: from typing import List, Any # To make HTTP requests: import httpx # To work with date and time: import datetime # For asynchronous activities: import asyncio # Misc: import math import random import string # ***************************************************************************************************************** # ***** **** # *** MACROS / ONE-TIME INIT *** # ***** **** # ***************************************************************************************************************** # Info for logging that will stay constant during runtime: SERVER_HOSTNAME = str(socket.gethostname()) PLATFORM_INFO = platform.uname() HOST_OS = str(distro.name(True)) HOST_CPU = f"{PLATFORM_INFO.processor} ({PLATFORM_INFO.machine})" # ***************************************************************************************************************** # ***** **** # *** VARIABLES *** # ***** **** # ***************************************************************************************************************** # --- Nothing Yet # ***************************************************************************************************************** # ***** **** # *** FUNCTIONS *** # ***** **** # ***************************************************************************************************************** # --- Nothing Yet # ***************************************************************************************************************** # ***** **** # *** CLASSES *** # ***** **** # ***************************************************************************************************************** class AsyncMikroTik: # ┏┓┓ ┓┏ # ┃ ┃┏┓┏┏ ┃┃┏┓┏┓┏ # ┗┛┗┗┻┛┛ ┗┛┗┻┛ ┛ ACTION_LOG_COLLECTION = "_mikrotikActions" # ┏┓ # ┃ ┏┓┏┓┏╋┏┓┓┏┏╋┏┓┏┓ # ┗┛┗┛┛┗┛┗┛ ┗┻┗┗┗┛┛ def __init__( self, config_by: str, mikrotik_ip: str, username: str, password: str, port: int = None, use_https: bool = True, http_client: httpx.AsyncClient = None, action_log_conn: AsyncMongo = None, debug: bool = True, debug_prefix: str = "MikroTik (C) | ", debug_only_errors: bool = True ): """ To control MikroTik devices in an async manner. :param http_client: The HTTP client to use to make REST-ful API calls. :param debug: Whether, or not, you would like to print debugging messages: :param debug_prefix: The prefix to print with the debugging messages. :param debug_only_errors: Whether you would like to print only error messages or all messages. :return: None. """ # Save the input values: self._config_by = config_by self._mikrotik_ip = mikrotik_ip self._username = username self._password = password self._port = port self._use_https = use_https # For logging actions to Mongo: self._action_log_conn = action_log_conn self._action_log_chain = None self.new_action_chain() # For controlled REST-ful calls: self._rest = AsyncREST( http_client = http_client, debug = debug, debug_prefix = debug_prefix, debug_only_errors = debug_only_errors ) self._basic_auth = BasicAuth( username = self._username, password = self._password ) # ┏┓ • # ┃┃┏┓┏┓┏┓┏┓┏┓╋┓┏┓┏ # ┣┛┛ ┗┛┣┛┗ ┛ ┗┗┗ ┛ # ┛ @property def action_chain(self): return self._action_log_chain # ┓┏ ┓ # ┣┫┏┓┃┏┓┏┓┏┓┏ # ┛┗┗ ┗┣┛┗ ┛ ┛ # ┛ def get_mikrotik_url( self, path: str, use_https: bool = None ) -> str: """ Simply creates the base URL for hitting the MikroTik server. :param path: The path of the REST API to hit. :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. :return: The Base URL string. """ use_https = self._use_https if use_https is None else use_https base_url = r"https://" if use_https else r"http://" base_url += self._mikrotik_ip if self._port is not None: base_url += f":{self._port}" base_url += "/rest" if not path.startswith("/"): path = "/" + path return base_url + path @staticmethod def split_ipv4_range_equally( start_ip: ipaddress.IPv4Address | str, end_ip: ipaddress.IPv4Address | str, targets: List ) -> List[dict]: """ Divides the IPv4 range over the list of targets (like VLAN ids) in a simple fashion :param start_ip: The first IP in the full pool (range). :param end_ip: The last IP in the full pool (range). :param targets: The list of targets (need not be sequential or ordered) like VLAN ids. :return: A list of dicts that describes the subnet for each target. """ # Parse the inputs: start_ip_obj = start_ip if isinstance(start_ip, ipaddress.IPv4Address) else ipaddress.IPv4Address(start_ip) end_ip_obj = end_ip if isinstance(end_ip, ipaddress.IPv4Address) else ipaddress.IPv4Address(end_ip) target_count = len(targets) # Calculate the total number of IPs in the range: total_ips = int(end_ip_obj) - int(start_ip_obj) + 1 # Figure out the no. of IPs in each block: ips_per_target = math.ceil(total_ips / target_count) # Iterate through the list of targets, and give them the IPs: subnets = [] current_ip = start_ip_obj for index, target in enumerate(targets): # Figure out the first and last IPs for this object: subnet_start_ip_obj = current_ip if index == target_count - 1: subnet_end_ip_obj = end_ip_obj else: subnet_end_ip_obj = subnet_start_ip_obj + ips_per_target - 1 # Add the calculated subnet to the list: subnets.append({ "target": target, "network": str(subnet_start_ip_obj) + "-" + str(subnet_end_ip_obj), "size": int(subnet_end_ip_obj) - int(subnet_start_ip_obj) + 1, "startIp": str(subnet_start_ip_obj), "endIp": str(subnet_end_ip_obj) }) # Update the current IP for the next subnet: current_ip += ips_per_target # Done here: return subnets @staticmethod def split_ipv4_range_in_powers_of_two( start_ip: ipaddress.IPv4Address | str, end_ip: ipaddress.IPv4Address | str, targets: List, consider_reserved_ips: bool = True, strict: bool = False ) -> List[dict]: """ Divides the IPv4 range over the list of targets (like VLAN ids) in blocks whose sizes are in powers of 2. This means that you won't have a block which has an unusual size like 7. :param start_ip: The first IP in the full pool (range). :param end_ip: The last IP in the full pool (range). :param targets: The list of targets (need not be sequential or ordered) like VLAN ids. :param consider_reserved_ips: Whether, or not, to add 3 IPs per block (network, gateway, and broadcast) when splitting the range. :param strict: Whether, or not, you want strict dvision of blocks such that the network address is always x.x.x.0 and the broadcast address is always x.x.x.255. :return: A list of dicts that describes the subnet for each target. """ # Parse the inputs: start_ip_obj = start_ip if isinstance(start_ip, ipaddress.IPv4Address) else ipaddress.IPv4Address(start_ip) end_ip_obj = end_ip if isinstance(end_ip, ipaddress.IPv4Address) else ipaddress.IPv4Address(end_ip) target_count = len(targets) # Calculate the total number of IPs in the range: total_ips = int(end_ip_obj) - int(start_ip_obj) + 1 # Calculate the no. of IPs each VLAN gets, # and add three (if asked) because we need IPs for network, gateway and broadcast: ips_per_target = math.ceil(total_ips / target_count) if consider_reserved_ips: ips_per_target += 3 # Generate subnets for each VLAN: subnets = [] current_ip = start_ip_obj for target in targets: # Calculate the network address for the current VLAN: subnet_network = ipaddress.IPv4Network(f"{current_ip}/{32 - ips_per_target.bit_length()}", strict = strict) # Check for overlap with the parent network: subnet_start_ip_obj = subnet_network.network_address subnet_end_ip_obj = subnet_network.broadcast_address if subnet_start_ip_obj <= end_ip_obj: # Add the calculated subnet to the list: subnets.append({ "target": target, "network": str(subnet_network), "size": int(2 ** (32 - subnet_network.prefixlen)), "startIp": str(subnet_start_ip_obj), "endIp": str(subnet_end_ip_obj) }) # Update the current IP for the next subnet: current_ip = ipaddress.IPv4Address(int(subnet_network.broadcast_address) + 1) # Done here: return subnets def create_comment_json( self, config_ts: datetime.datetime = None, roll_back_config: dict = None ) -> str: """ To create a JSON string that can be put as a comment in any step to later identify the work that was done and have some hint about how to roll it back if needed. :param config_ts: The datetime (preferably UTC) when this comment was created. :param roll_back_config: The original config that can be used when rolling back. :return: The comment string to be used. """ # Create the JSON: comment_json = { "configBy": self._config_by, "configTs": (config_ts or date_time.get_current_utc_date_time(as_string = False)).timestamp(), "rollbackConfig": roll_back_config } # Return a string: return json.to_string(comment_json, no_space = True) @staticmethod def parse_comment_json(comment: str) -> dict | list | None: """ Tries to parse a comment as if it were a JSON string. :param comment: The comment string. :return: The dict/list parsed form the JSON, or a null value of the input was not a valid JSON string. """ # Start with a null value: parsed = None # Try to parse the comment as if it were a JSON string: try: parsed = json.from_string(comment) except: parsed = None # Done here: return parsed @staticmethod def get_original_config(comment_json: dict) -> dict | None: """ Once you have the comment's JSON unpacked successfully, pass that dict here to get the original config back. :param comment_json: The dict received from parsing the comment's JSON string. :return: The dict that describes the previous config, or None if no previous config was found. """ return comment_json.get("rollbackConfig") if isinstance(comment_json, dict) else None def is_my_config( self, config_json: dict, name_substring: str = None ) -> bool: """ Looks at a particular resource's JSON and tells if it was created by this class. :param config_json: The JSON of the resource whose config needs to be checked. :param name_substring: A substring (case-sensitive) to look for in the 'name' field. :return: True if the config was created by this class, else False. """ # Extract needed values: name = config_json.get("name", "???") comment_json = self.parse_comment_json(config_json.get("comment")) or {} # Perform needed checks: if ( name.find(name_substring or "bhopli-muchhi-cookie") >= 0 or comment_json.get("configBy", "???") == self._config_by ): return True else: return False def new_action_chain(self) -> None: """ Generates a new random string to mark a new action chain and saves it in the local variable. :return: None """ self._action_log_chain = "".join(random.choices(string.ascii_letters + string.digits, k = 8)) async def log_action( self, api_response: ApiResponse, ) -> None: """ Invoke this to record actions automatically. :param api_response: The response received from invoking an API provided by MikroTik. :return: None. """ # Proceed only if we have a logging connection: if self._action_log_conn is not None: # Prepare the final log: action_log_json = { "hostname": SERVER_HOSTNAME, "hostOs": HOST_OS, "hostCpu": HOST_CPU, "actionChain": self._action_log_chain, "configBy": self._config_by, "mikrotikIp": self._mikrotik_ip, "url": api_response.url, "method": api_response.method, "action": api_response.action, "config": api_response.requestJson, "httpCode": api_response.httpCode, "success": api_response.success, "message": api_response.message, "exception": describe_exception(api_response.exception), "ts": date_time.get_current_utc_date_time(as_string = False) } # Record the action: asyncio.create_task(self._action_log_conn.insert_one( collection = self.ACTION_LOG_COLLECTION, document = action_log_json )) @staticmethod async def get_failure_message(api_response: ApiResponse) -> str: # On successful API calls, MikroTik doesn't give messages: if api_response.success: message = "Success" # On failure, the message is available in the JSON payload: else: response_json = await api_response.get_json() message = f"{response_json.get('message')} ({response_json.get('detail')})" # Done here: return message # ┏┓ # ┗┓┓┏┏╋┏┓┏┳┓ # ┗┛┗┫┛┗┗ ┛┗┗ # ┛ async def list_system_resources( self, use_https: bool = None ) -> ApiResponse: """ To get a summary of the hardware resources available in the MikroTik device. This also becomes a great way to quickly check if any given device is valid, and up and running. :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. Leave it as null to use the class's default setting. :return: A structured API response. """ # Make the API call: api_response = await self._rest.get( url = self.get_mikrotik_url(path = r"/system/resource", use_https = use_https), auth = self._basic_auth ) # Extract the needed values: if api_response.success: api_response.data = await api_response.get_json() # Done here: await self.log_action(api_response) return api_response # ┳ ┏ # ┃┏┓╋┏┓┏┓╋┏┓┏┏┓┏ # ┻┛┗┗┗ ┛ ┛┗┻┗┗ ┛ async def list_interfaces( self, use_https: bool = None ) -> ApiResponse: """ To enlist all the physical connectivity interfaces available on the MikroTik device. :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. Leave it as null to use the class's default setting. :return: A structured API response. """ # Make the API call: api_response = await self._rest.get( url = self.get_mikrotik_url(path = r"/interface", use_https = use_https), auth = self._basic_auth ) # Extract the needed values: if api_response.success: api_response.data = await api_response.get_json() else: api_response.message = await self.get_failure_message(api_response) # Done here: await self.log_action(api_response) return api_response async def update_interface( self, dot_id: str, json_payload: dict, use_https: bool = None ) -> ApiResponse: """ To update an interface's configuration. :param dot_id: The value of the '.id' field in the list. :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. :param json_payload: The JSON to send in the body pf the request. :return: A structured API response. """ # Make the API call: api_response = await self._rest.patch( url = self.get_mikrotik_url( path = f"/interface/{dot_id}", use_https = use_https ), auth = self._basic_auth, json = json_payload ) # Extract the needed values: if api_response.success: api_response.data = await api_response.get_json() else: api_response.message = await self.get_failure_message(api_response) # Done here: await self.log_action(api_response) return api_response # ┓┏┓ ┏┓┳┓ # ┃┃┃ ┣┫┃┃┏ # ┗┛┗┛┛┗┛┗┛ async def list_vlans( self, use_https: bool = None ) -> ApiResponse: """ To enlist all the physical connectivity interfaces available on the MikroTik device. :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. :return: A structured API response. """ # Make the API call: api_response = await self._rest.get( url = self.get_mikrotik_url(path = r"/interface/vlan", use_https = use_https), auth = self._basic_auth ) # Extract the needed values: if api_response.success: api_response.data = await api_response.get_json() else: api_response.message = await self.get_failure_message(api_response) # Done here: await self.log_action(api_response) return api_response async def add_vlan( self, json_payload: dict, use_https: bool = None ) -> ApiResponse: """ To add a new VLAN to the MikroTik's configuration. :param json_payload: The JSON to send in the body pf the request. :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. :return: A structured API response. """ # Make the API call: api_response = await self._rest.put( url = self.get_mikrotik_url(path = r"/interface/vlan", use_https = use_https), auth = self._basic_auth, json = json_payload ) # Extract the needed values: if api_response.success: api_response.data = await api_response.get_json() else: api_response.message = await self.get_failure_message(api_response) # Done here: await self.log_action(api_response) return api_response async def update_vlan( self, dot_id: str, json_payload: dict, use_https: bool = None, ) -> ApiResponse: """ To update an existing VLAN in the MikroTik's configuration. :param dot_id: The value of the '.id' field in the list. :param json_payload: The JSON to send in the body pf the request. :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. :return: A structured API response. """ # Make the API call: api_response = await self._rest.patch( url = self.get_mikrotik_url(path = f"/interface/vlan/{dot_id}", use_https = use_https), auth = self._basic_auth, json = json_payload ) # Extract the needed values: if api_response.success: api_response.data = await api_response.get_json() else: api_response.message = await self.get_failure_message(api_response) # Done here: await self.log_action(api_response) return api_response async def remove_vlan( self, dot_id: str, use_https: bool = None, ): """ To delete an existing VLAN from the MikroTik's configuration. :param dot_id: The value of the '.id' field in the list. :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. :return: A structured API response. """ # Make the API call: api_response = await self._rest.delete( url = self.get_mikrotik_url(path = f"/interface/vlan/{dot_id}", use_https = use_https), auth = self._basic_auth ) # Extract the needed values: if api_response.success: api_response.data = await api_response.get_json() else: api_response.message = await self.get_failure_message(api_response) # Done here: await self.log_action(api_response) return api_response # ┳┏┓ ┏┓ ┓ # ┃┃┃ ┃┃┏┓┏┓┃┏ # ┻┣┛ ┣┛┗┛┗┛┗┛ async def list_ip_pools( self, use_https: bool = None ) -> ApiResponse: """ To enlist all the IP Pools that have been configured. :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. :return: A structured API response. """ # Make the API call: api_response = await self._rest.get( url = self.get_mikrotik_url(path = r"/ip/pool", use_https = use_https), auth = self._basic_auth ) # Extract the needed values: if api_response.success: api_response.data = await api_response.get_json() else: api_response.message = await self.get_failure_message(api_response) # Done here: await self.log_action(api_response) return api_response async def add_ip_pool( self, json_payload: dict, use_https: bool = None ) -> ApiResponse: """ To add a new IP Pool to the MikroTik's configuration. :param json_payload: The JSON to send in the body pf the request. :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. :return: A structured API response. """ # Make the API call: api_response = await self._rest.put( url = self.get_mikrotik_url(path = r"/ip/pool", use_https = use_https), auth = self._basic_auth, json = json_payload ) # Extract the needed values: if api_response.success: api_response.data = await api_response.get_json() else: api_response.message = await self.get_failure_message(api_response) # Done here: await self.log_action(api_response) return api_response async def update_ip_pool( self, dot_id: str, json_payload: dict, use_https: bool = None, ) -> ApiResponse: """ To update an existing IP Pool in the MikroTik's configuration. :param dot_id: The value of the '.id' field in the list. :param json_payload: The JSON to send in the body pf the request. :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. :return: A structured API response. """ # Make the API call: api_response = await self._rest.patch( url = self.get_mikrotik_url(path = f"/ip/pool/{dot_id}", use_https = use_https), auth = self._basic_auth, json = json_payload ) # Extract the needed values: if api_response.success: api_response.data = await api_response.get_json() else: api_response.message = await self.get_failure_message(api_response) # Done here: await self.log_action(api_response) return api_response async def remove_ip_pool( self, dot_id: str, use_https: bool = None, ): """ To delete an existing IP Pool from the MikroTik's configuration. :param dot_id: The value of the '.id' field in the list. :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. :return: A structured API response. """ # Make the API call: api_response = await self._rest.delete( url = self.get_mikrotik_url(path = f"/ip/pool/{dot_id}", use_https = use_https), auth = self._basic_auth ) # Extract the needed values: if api_response.success: api_response.data = await api_response.get_json() else: api_response.message = await self.get_failure_message(api_response) # Done here: await self.log_action(api_response) return api_response # ┏┓┏┓┏┓ ┏┓ ┏•┓ # ┃┃┃┃┃┃ ┃┃┏┓┏┓╋┓┃┏┓ # ┣┛┣┛┣┛ ┣┛┛ ┗┛┛┗┗┗ async def list_ppp_profiles( self, use_https: bool = None ) -> ApiResponse: """ To enlist all the PPP Profiles that have been configured. :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. :return: A structured API response. """ # Make the API call: api_response = await self._rest.get( url = self.get_mikrotik_url(path = r"/ppp/profile", use_https = use_https), auth = self._basic_auth ) # Extract the needed values: if api_response.success: api_response.data = await api_response.get_json() else: api_response.message = await self.get_failure_message(api_response) # Done here: await self.log_action(api_response) return api_response async def add_ppp_profile( self, json_payload: dict, use_https: bool = None ) -> ApiResponse: """ To add a new PPP Profile to the MikroTik's configuration. :param json_payload: The JSON to send in the body pf the request. :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. :return: A structured API response. """ # Make the API call: api_response = await self._rest.put( url = self.get_mikrotik_url(path = r"/ppp/profile", use_https = use_https), auth = self._basic_auth, json = json_payload ) # Extract the needed values: if api_response.success: api_response.data = await api_response.get_json() else: api_response.message = await self.get_failure_message(api_response) # Done here: await self.log_action(api_response) return api_response async def update_ppp_profile( self, dot_id: str, json_payload: dict, use_https: bool = None, ) -> ApiResponse: """ To update an existing PPP Profile in the MikroTik's configuration. :param dot_id: The value of the '.id' field in the list. :param json_payload: The JSON to send in the body pf the request. :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. :return: A structured API response. """ # Make the API call: api_response = await self._rest.patch( url = self.get_mikrotik_url(path = f"/ppp/profile/{dot_id}", use_https = use_https), auth = self._basic_auth, json = json_payload ) # Extract the needed values: if api_response.success: api_response.data = await api_response.get_json() else: api_response.message = await self.get_failure_message(api_response) # Done here: await self.log_action(api_response) return api_response async def remove_ppp_profile( self, dot_id: str, use_https: bool = None, ): """ To delete an existing PPP Profile from the MikroTik's configuration. :param dot_id: The value of the '.id' field in the list. :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. :return: A structured API response. """ # Make the API call: api_response = await self._rest.delete( url = self.get_mikrotik_url(path = f"/ppp/profile/{dot_id}", use_https = use_https), auth = self._basic_auth ) # Extract the needed values: if api_response.success: api_response.data = await api_response.get_json() else: api_response.message = await self.get_failure_message(api_response) # Done here: await self.log_action(api_response) return api_response # ┏┓┏┓┏┓ ┏┓ # ┃┃┃┃┃┃ ┗┓┏┓┏┓┓┏┏┓┏┓ # ┣┛┣┛┣┛ ┗┛┗ ┛ ┗┛┗ ┛ async def list_ppp_servers( self, use_https: bool = None ) -> ApiResponse: """ To enlist all the PPP Servers that have been configured. :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. :return: A structured API response. """ # Make the API call: api_response = await self._rest.get( url = self.get_mikrotik_url(path = r"/interface/pppoe-server/server", use_https = use_https), auth = self._basic_auth ) # Extract the needed values: if api_response.success: api_response.data = await api_response.get_json() else: api_response.message = await self.get_failure_message(api_response) # Done here: await self.log_action(api_response) return api_response async def add_ppp_server( self, json_payload: dict, use_https: bool = None ) -> ApiResponse: """ To add a new PPP Server to the MikroTik's configuration. :param json_payload: The JSON to send in the body pf the request. :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. :return: A structured API response. """ # Make the API call: api_response = await self._rest.put( url = self.get_mikrotik_url(path = r"/interface/pppoe-server/server", use_https = use_https), auth = self._basic_auth, json = json_payload ) # Extract the needed values: if api_response.success: api_response.data = await api_response.get_json() else: api_response.message = await self.get_failure_message(api_response) # Done here: await self.log_action(api_response) return api_response async def update_ppp_server( self, dot_id: str, json_payload: dict, use_https: bool = None, ) -> ApiResponse: """ To update an existing PPP Server in the MikroTik's configuration. :param dot_id: The value of the '.id' field in the list. :param json_payload: The JSON to send in the body pf the request. :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. :return: A structured API response. """ # Make the API call: api_response = await self._rest.patch( url = self.get_mikrotik_url(path = f"/interface/pppoe-server/server/{dot_id}", use_https = use_https), auth = self._basic_auth, json = json_payload ) # Extract the needed values: if api_response.success: api_response.data = await api_response.get_json() else: api_response.message = await self.get_failure_message(api_response) # Done here: await self.log_action(api_response) return api_response async def remove_ppp_server( self, dot_id: str, use_https: bool = None, ): """ To delete an existing PPP Server from the MikroTik's configuration. :param dot_id: The value of the '.id' field in the list. :param use_https: Whether to use HTTPS, or HTTP. Leave it as null to use the class's default setting. :return: A structured API response. """ # Make the API call: api_response = await self._rest.delete( url = self.get_mikrotik_url(path = f"/interface/pppoe-server/server/{dot_id}", use_https = use_https), auth = self._basic_auth ) # Extract the needed values: if api_response.success: api_response.data = await api_response.get_json() else: api_response.message = await self.get_failure_message(api_response) # Done here: await self.log_action(api_response) return api_response # ***************************************************************************************************************** # ***** **** # *** MAIN PROGRAM *** # ***** **** # ***************************************************************************************************************** if __name__ == "__main__": # Make an instance of an HTTP client to use to make API calls: my_http_client = httpx.AsyncClient( limits = httpx.Limits( max_connections = 100, # ............ Maximum number of connections allowed in the pool. max_keepalive_connections = 50, # ... Maximum number of connections that can be kept alive. ), timeout = httpx.Timeout( pool = 120.0, # .... Time to wait for a free connection from the pool. connect = 2.5, # ... Time to wait for establishing a connection to the server. write = 10.0, # .... Time to wait for sending data. read = 9.9 # ....... Time to wait for receiving data. ) ) # Create an instance of a Mongo conn. to write action logs: logs_mongo = AsyncMongo( connection_string = r"mongodb://del.ditscentre.in:27017,wtt.ditscentre.in:27017,mum.arh.001.ditscentre.in:27017/admin?tls=true&tlsCAFile=%2Fetc%2Fssl%2Fdbu%2Fmongo_data_ca.pem&tlsCertificateKeyFile=%2Fetc%2Fssl%2Fdbu%2Fmongo_data_cert.pem&replicaSet=dits_mongod_rep&readPreference=primary&authMechanism=MONGODB-X509&authSource=%24external", database_name = "converse", max_connections = 10, ) # Create an instance of a MikroTik client: my_mikrotik = AsyncMikroTik( http_client = my_http_client, mikrotik_ip = "102.210.173.150", username = "easyfi", password = "easyfi", use_https = False, config_by = "bhopli", action_log_conn = logs_mongo ) async def main(): await logs_mongo.connect() # api_response = await my_mikrotik.list_system_resources() # print("SYS RESOURCES:", json.to_string(api_response.data)) # # api_response = await my_mikrotik.list_interfaces() # print(f"INTERFACES ({len(api_response.data)}):", json.to_string(api_response.data)) # # api_response = await my_mikrotik.list_vlans() # print(f"VLANS (({len(api_response.data)})):", json.to_string(api_response.data)) api_response = await my_mikrotik.list_ip_pools() print(f"IP POOLS ({len(api_response.data)}):", json.to_string(api_response.data)) # api_response = await my_mikrotik.list_ppp_profiles() # print(f"PPP PROFILES ({len(api_response.data)}):", json.to_string(api_response.data)) # # api_response = await my_mikrotik.list_ppp_servers() # print(f"PPP SERVERS ({len(api_response.data)}):", json.to_string(api_response.data)) asyncio.run(main())