From b8b334d5e36d5ae4317eb4f7251ae95dcd76ce6f Mon Sep 17 00:00:00 2001 From: dormant-user Date: Wed, 4 Sep 2024 10:06:16 -0500 Subject: [PATCH] Add more routes to get IP, CPU and memory info --- docs/genindex.html | 22 +++++++- docs/index.html | 131 ++++++++++++++++++++++++++++++++++++++++++++ docs/objects.inv | Bin 1015 -> 1075 bytes docs/searchindex.js | 2 +- pyninja/routers.py | 123 ++++++++++++++++++++++++++++++++++++++--- pyninja/squire.py | 82 +++++++++++++++++++++++++++ 6 files changed, 347 insertions(+), 13 deletions(-) diff --git a/docs/genindex.html b/docs/genindex.html index 72fd3ed..8badda0 100644 --- a/docs/genindex.html +++ b/docs/genindex.html @@ -139,10 +139,12 @@

F

@@ -278,10 +288,14 @@

P

  • parse_service_manager() (pyninja.models.EnvConfig class method)
  • Payload (class in pyninja.models) +
    • +
  • private_ip_address() (in module pyninja.squire)
  • process_command() (in module pyninja.squire)
  • process_status() (in module pyninja.routers) +
    • +
  • public_ip_address() (in module pyninja.squire)
  • put_record() (in module pyninja.database)
    • @@ -402,11 +416,13 @@

    S

  • ServiceManager (class in pyninja.models)
    • - - +
    • Session (class in pyninja.models) +
      • +
    • size_converter() (in module pyninja.squire)
    • start() (in module pyninja.main)
      • diff --git a/docs/index.html b/docs/index.html index a77c2db..936b9b1 100644 --- a/docs/index.html +++ b/docs/index.html @@ -185,6 +185,61 @@

      Welcome to PyNinja’s documentation!

      Routers

      +
      +
      +async pyninja.routers.get_ip(request: Request, public: bool = False, apikey: HTTPAuthorizationCredentials = Depends(HTTPBearer))
      +

      Get local and public IP address of the device.

      +
      +
      Parameters:
      +
        +

      • request – Reference to the FastAPI request object.

        • +

      • public – Boolean flag to get the public IP address.

        • +

      • apikey – API Key to authenticate the request.

        • +

      • token – API secret to authenticate the request.

        • +
      +
      +
      +

      Raises:

      +
      +

      APIResponse: +Raises the HTTPStatus object with a status code and the public/private IP as response.

      +
      +
      + +
      +
      +async pyninja.routers.get_cpu(request: Request, interval: int | float = 2, per_cpu: bool = True, apikey: HTTPAuthorizationCredentials = Depends(HTTPBearer))
      +

      Get the CPU utilization.

      +

      Args:

      +
      +

      request: Reference to the FastAPI request object. +interval: Interval to get the CPU utilization. +per_cpu: If True, returns the CPU utilization for each CPU. +apikey: API Key to authenticate the request.

      +
      +

      Raises:

      +
      +

      APIResponse: +Raises the HTTPStatus object with a status code and CPU usage as response.

      +
      +
      + +
      +
      +async pyninja.routers.get_memory(request: Request, apikey: HTTPAuthorizationCredentials = Depends(HTTPBearer))
      +

      Get memory utilization.

      +

      Args:

      +
      +

      request: Reference to the FastAPI request object. +apikey: API Key to authenticate the request.

      +
      +

      Raises:

      +
      +

      APIResponse: +Raises the HTTPStatus object with a status code and CPU usage as response.

      +
      +
      +
      async pyninja.routers.run_command(request: Request, payload: Payload, apikey: HTTPAuthorizationCredentials = Depends(HTTPBearer), token: Optional[str] = Header(None))
      @@ -303,6 +358,20 @@

      Welcome to PyNinja’s documentation! +
      +async pyninja.routers.health()
      +

      Health check for PyNinja.

      +
      +
      Returns:
      +

      Returns a health check response with status code 200.

      +
      +
      Return type:
      +

      APIResponse

      +
      +
      +

      +
      pyninja.routers.get_all_routes() → List[APIRoute]
      @@ -928,6 +997,68 @@

      Models

      Squire

      +
      +
      +pyninja.squire.public_ip_address() → str
      +

      Gets public IP address of the host using different endpoints.

      +
      +
      Returns:
      +

      Public IP address.

      +
      +
      Return type:
      +

      str

      +
      +
      +
      + +
      +
      +pyninja.squire.private_ip_address() → str | None
      +

      Uses a simple check on network id to see if it is connected to local host or not.

      +
      +
      Returns:
      +

      Private IP address of host machine.

      +
      +
      Return type:
      +

      str

      +
      +
      +
      + +
      +
      +pyninja.squire.format_nos(input_: float) → int | float
      +

      Removes .0 float values.

      +
      +
      Parameters:
      +

      input_ – Strings or integers with .0 at the end.

      +
      +
      Returns:
      +

      Int if found, else returns the received float value.

      +
      +
      Return type:
      +

      int | float

      +
      +
      +
      + +
      +
      +pyninja.squire.size_converter(byte_size: int | float) → str
      +

      Gets the current memory consumed and converts it to human friendly format.

      +
      +
      Parameters:
      +

      byte_size – Receives byte size as argument.

      +
      +
      Returns:
      +

      Converted understandable size.

      +
      +
      Return type:
      +

      str

      +
      +
      +
      +
      pyninja.squire.process_command(command: str, timeout: Union[int, float]) → Dict[str, List[str]]
      diff --git a/docs/objects.inv b/docs/objects.inv index e2ad5820514acf937ef5af047b5c279b90fbf999..ab3efa4d10e9f30283c7d5f243e86983fdcfdad5 100644 GIT binary patch delta 960 zcmV;x13&!t2eSx}fq&ZT3fUfW3z7vAAlnRfQ1mEhEzVe7SyJ>fKRNX^@_KoaQkFeG z$pT|)Znn&iq$r6Z?G8qne+3gMc8xo{?X652nm3Iy685Z%z5cz7-Tl$rE!>$ndsu#X z|1m@BWZI~#Y3|RlU*y-;KPxH05SUIFse$Kb4&GZ&V3lb-^nYL?P$&&s(EOfE{5g-1 zB8q1bfDr5nVH=3tLuid59vWIwd|C9N&;b0s9tQ772tUI6>^)^3xLphmXTHJ${*Ek^OB} z$vNvI$S?L9jylB7 z%B%c?*}k`Ctv2&b9uamciK`gB$h?TrtLGGfuA@^DxH388C{t2)s8uPP;sG#}@dgbq zaHyrvb9D{B_O|C>wtTJXb^+6NIQ`?Y*7z_C^Uj8Nn}3FUqH8rc;RD{`7pqAQMNtyK zZ&eGt7Y1{?uhQi6$<(S>)f39sBI=A3Vgy)g zd-PXUSz}T4U=uO$Pq<*8r}diW>=D7sVvqI9*90MkO0&35cnBpCUzNpuJgx0K*DpD| zQv6PtTYpTm&&s6zO&ZVxX?Jx2SRD%>()RGg9a4_Y^jPrt1C8rI=<-u(K2voj zMi#qeepZ{z2dBhAF*-7Yy*&~asj9*v8-EJWpb)KdAZbuCVk?xMT38DBV(f0_Z@%y~ zDSuB2oB^y6Jaz{`OkZS09n z&sw#$mV^2RIRkIOU;Tml^ziP(<3kgoe2Yo2xw~)5$RN=y4}WFFiw&588{}|XfzFcC zXp#Mt9~URFQQeuuKcF!f&_p{a{_)@U&^e%B%Kv&r_VBj_?weO^xx?M`p(uMSTdr6= i`+5ai9`NQfmvjK(+wKx9V})6j?@PRs7XJg)Aa7sXlh_9U delta 899 zcmV-}1AP3m2=@n&fq&BL3f&%a3z7vIAe#(!Q1mEhEzXR(vZUx|etPO_^!55Ar7U}X z(gjx3+-#X2Nl_vt+8vEDe+nj2+;;Bxey}oW=zi&xk#Jz!IOum}?C$68Zs9J(*~9YN z$KNuvPNt8_n(qD*`=9*U`e!927y`2uMrz>snS=M%6If+>4}Sxg2oy@g4m7_f6Tj9G zQbh4A0uX{dBWwead+4o~G+WH2MiAp?cR*up{{mk91(Gtw6EsdzXw7?TqEH5WpxQmD zP=nY&JHE3z4UPQCdmf^QsZEJb2FEw#P{5%Fmqh|$3CAZ4!Qtb$O;2BHEVAFt zDmiC;1o>@MiGL3`JA$iOoMBvjtcwS#08NQS0obRDTbRAnIfh&_MjxwcWhgOxsDINer8E??= z0>@VRvR1e7Yi|b*X2;j6Z5J?Y$Lyb$wZ(^Vm~U-}w|{BK8QrSE2_NtbzgkUlD2kE* zey4iigD_azeUm1iGgGTUHBTrXi>Lz*(5JEVH0pU}HZ5xQ*1rHX&6eV*t&sKQVggui z2lQ7qSz}QRU=uO$Pq<)T=Jl#`_JrVJalre__XHt^Mzgq1n1qsuugc;+p4Ya{^=l4~ z6u(#I7Jt+1i!$kOlLqua+I?F9R;LRPY5oOc5<>5lqi5N*nsN)nS}K96CWOg$t@|Uu z@CIr3Ai4EwqfBVBI`$Xnbeek@x`xtpwXw>qZ5!zwal%+soQ8vnN8WEC--0PuyYaUC zN-=$#ci#5sGvRJH5u-^59^;D!dmx!BaCs&*2!Gc{SEa{i=v>mO3exK!ZVOd)gnKOX z)3H?RX(6UQh}`%5fK%GPA5c>3cfk^gxzilqiEPvMnOJNr`Nc6_56(G8F*-4XZ7>m6 zS>0ffP5m{mZ_TPL_Vn=K)8j)IqI{1D)ZN{8Wn`4-mdCGI^I`*Ll@heh*i^E Z(*cC RedirectResponse: return RedirectResponse("/docs") -async def health() -> exceptions.APIResponse: +async def health(): """Health check for PyNinja. Returns: @@ -257,7 +342,27 @@ def get_all_routes() -> List[APIRoute]: ] routes = [ APIRoute(path="/", endpoint=docs, methods=["GET"], include_in_schema=False), - APIRoute(path="/health", endpoint=health, methods=["GET"], include_in_schema=False), + APIRoute( + path="/health", endpoint=health, methods=["GET"], include_in_schema=False + ), + APIRoute( + path="/get-ip", + endpoint=get_ip, + methods=["GET"], + dependencies=dependencies, + ), + APIRoute( + path="/get-cpu", + endpoint=get_cpu, + methods=["GET"], + dependencies=dependencies, + ), + APIRoute( + path="/get-memory", + endpoint=get_memory, + methods=["GET"], + dependencies=dependencies, + ), APIRoute( path="/service-status", endpoint=service_status, diff --git a/pyninja/squire.py b/pyninja/squire.py index 54e1a79..e9afed9 100644 --- a/pyninja/squire.py +++ b/pyninja/squire.py @@ -1,16 +1,98 @@ import json import logging +import math import os import pathlib +import re +import socket import subprocess from typing import Dict, List +import requests import yaml from pydantic import PositiveFloat, PositiveInt from pyninja.models import EnvConfig LOGGER = logging.getLogger("uvicorn.default") +IP_REGEX = re.compile( + r"""^(?:(?:25[0-5]|2[0-4][0-9]|1[0-9][0-9]|[1-9][0-9]|[0-9])\.){3}(?:25[0-5]|2[0-4][0-9]|1[0-9][0-9]|[1-9][0-9]|[0-9])$""" # noqa: E501 +) + + +def public_ip_address() -> str: + """Gets public IP address of the host using different endpoints. + + Returns: + str: + Public IP address. + """ + fn1 = lambda fa: fa.text.strip() # noqa: E731 + fn2 = lambda fa: fa.json()["origin"].strip() # noqa: E731 + mapping = { + "https://checkip.amazonaws.com/": fn1, + "https://api.ipify.org/": fn1, + "https://ipinfo.io/ip/": fn1, + "https://v4.ident.me/": fn1, + "https://httpbin.org/ip": fn2, + "https://myip.dnsomatic.com/": fn1, + } + for url, func in mapping.items(): + try: + with requests.get(url) as response: + return IP_REGEX.findall(func(response))[0] + except ( + requests.RequestException, + requests.JSONDecodeError, + re.error, + IndexError, + ): + continue + + +def private_ip_address() -> str | None: + """Uses a simple check on network id to see if it is connected to local host or not. + + Returns: + str: + Private IP address of host machine. + """ + socket_ = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + try: + socket_.connect(("8.8.8.8", 80)) + except OSError: + return + ip_address_ = socket_.getsockname()[0] + socket_.close() + return ip_address_ + + +def format_nos(input_: float) -> int | float: + """Removes ``.0`` float values. + + Args: + input_: Strings or integers with ``.0`` at the end. + + Returns: + int | float: + Int if found, else returns the received float value. + """ + return int(input_) if isinstance(input_, float) and input_.is_integer() else input_ + + +def size_converter(byte_size: int | float) -> str: + """Gets the current memory consumed and converts it to human friendly format. + + Args: + byte_size: Receives byte size as argument. + + Returns: + str: + Converted understandable size. + """ + size_name = ("B", "KB", "MB", "GB", "TB", "PB", "EB", "ZB", "YB") + index = int(math.floor(math.log(byte_size, 1024))) + return f"{format_nos(round(byte_size / pow(1024, index), 2))} {size_name[index]}" def process_command(