Implement basic AWS client functions

.pre-commit-config.yaml

@@ -44,8 +44,9 @@ repos:
     rev: v1.1.325
   - hooks:
       - id: mypy
-        # exclude: (^docs/|^tests/)
-        # files: (^aws_pi_reports/)
+        exclude: (^docs/|^tests/)
+        files: (^aws_pi_reports/)
+        # args: ["--verbose"]
     repo: https://github.com/pre-commit/mirrors-mypy
     rev: v1.5.1
 

aws_pi_reports/aws.py

@@ -0,0 +1,102 @@
+""""AWS boto functions for pi_reports"""
+
+from datetime import datetime
+from typing import Any, List, Literal
+
+from boto3.session import Session
+from mypy_boto3_pi.client import PIClient
+from mypy_boto3_pi.type_defs import GetResourceMetricsResponseTypeDef, MetricQueryTypeDef
+from mypy_boto3_rds.client import RDSClient
+from mypy_boto3_rds.type_defs import DBInstanceMessageTypeDef
+
+from aws_pi_reports.time import parse_time
+
+
+class PIAwsClient:
+    """PI AWS Client"""
+
+    def __init__(self, session: Session) -> None:
+        self._session: Session = session
+        self._pi_client: PIClient = self._session.client("pi")  # pyright: ignore[reportUnknownMemberType]
+
+    def pi_get_resource_metrics(
+        self,
+        service_type: Literal["DOCDB", "RDS"],
+        identifier: str,
+        metric_queries: List[MetricQueryTypeDef],
+        start_time: datetime,
+        end_time: datetime,
+        period_in_seconds: int = 3600,
+        max_results: int = 100,
+        next_token: str = "",
+        period_alignment: Literal["END_TIME", "START_TIME"] = "END_TIME",
+    ) -> GetResourceMetricsResponseTypeDef:
+        result: GetResourceMetricsResponseTypeDef = self._pi_client.get_resource_metrics(
+            ServiceType=service_type,
+            Identifier=identifier,
+            MetricQueries=metric_queries,
+            StartTime=start_time,
+            EndTime=end_time,
+            PeriodInSeconds=period_in_seconds,
+            MaxResults=max_results,
+            NextToken=next_token,
+            PeriodAlignment=period_alignment,
+        )
+        return result
+
+
+class RDSAwsCClient:
+    """RDS AWS Client"""
+
+    def __init__(
+        self,
+        session: Session,
+    ) -> None:
+        self._session: Session = session
+        self._rds_client: RDSClient = self._session.client("rds")  # pyright: ignore[reportUnknownMemberType]
+
+    def _rds_get_attribute(self, db_instance_identifier: str, attr_name: str) -> Any:
+        response: DBInstanceMessageTypeDef = self._rds_client.describe_db_instances(DBInstanceIdentifier=db_instance_identifier)
+        return response["DBInstances"][0][attr_name]  # type: ignore # pyright: ignore[reportUnknownVariableType]
+
+    def rds_get_database_instance_resource_id(self, db_instance_identifier: str) -> str:
+        return str(self._rds_get_attribute(db_instance_identifier=db_instance_identifier, attr_name="DbiResourceId"))
+
+
+class AWSClient(PIAwsClient, RDSAwsCClient):
+    """AWS Client"""
+
+    def __init__(self, aws_profile: str = "", aws_region: str = "") -> None:
+        self._session: Session = Session(profile_name=aws_profile, region_name=aws_region)
+        super().__init__(session=self._session)  # Initialize all supers with session
+
+    def get_resource_metrics_for_db_instance(
+        self,
+        db_instance_identifier: str,
+        service_type: Literal["DOCDB", "RDS"],
+        metric_queries: List[MetricQueryTypeDef],
+        time: datetime,
+        time_delta: str,
+        period_in_seconds: int = 3600,
+        max_results: int = 100,
+        next_token: str = "",
+        period_alignment: Literal["END_TIME", "START_TIME"] = "END_TIME",
+    ) -> GetResourceMetricsResponseTypeDef:
+        start_time, end_time = parse_time(time, time_delta)
+
+        if service_type == "RDS":
+            resource_identifier: str = self.rds_get_database_instance_resource_id(db_instance_identifier=db_instance_identifier)
+        else:
+            raise NotImplementedError(f"Service type {service_type} not implemented")
+
+        return self.pi_get_resource_metrics(
+            service_type=service_type,
+            identifier=resource_identifier,
+            metric_queries=metric_queries,
+            start_time=start_time,
+            end_time=end_time,
+            period_in_seconds=period_in_seconds,
+            max_results=max_results,
+            next_token=next_token,
+            period_alignment=period_alignment,
+        )

aws_pi_reports/rds/cli.py

@@ -1,9 +1,18 @@
 """cli for RDS reports"""
-
+import inspect
+from types import FrameType
+from typing import Union
 
 import typer
 
 app = typer.Typer(
     help="""Performance Insights Reports for RDS
     """
 )
+
+
+@app.command()
+def counter_metrics() -> None:
+    frame: Union[FrameType, None] = inspect.currentframe()
+    f_name = frame.f_code.co_name if frame else "unknown_function"
+    print(f"Executing {f_name}")

aws_pi_reports/reports.py

@@ -0,0 +1,37 @@
+"""Reports module"""
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+
+
+class Report(ABC):
+    """
+    Base Report interface with methods to be implemented by concrete reports
+    """
+
+    # @property
+    # @abstractmethod
+    # def product(self) -> None:
+    #     pass
+
+    @abstractmethod
+    def read_report_input(self) -> None:
+        pass
+
+    @abstractmethod
+    def processs_report_input(self) -> None:
+        pass
+
+    @abstractmethod
+    def processs_queries_output(self) -> None:
+        pass
+
+    @abstractmethod
+    def report(self) -> None:
+        pass
+
+
+class StandardRDSReport(Report):
+    """
+    Standard RDS Report
+    """

aws_pi_reports/reports_inputs/counter_metrics.json

@@ -0,0 +1,10 @@
+{
+    "metric-queries":[
+       {
+          "Metric":"os.cpuUtilization.user.avg"
+       },
+       {
+          "Metric":"os.cpuUtilization.idle.avg"
+       }
+    ]
+ }

aws_pi_reports/reports_inputs/reference.txt

@@ -0,0 +1 @@
+https://docs.aws.amazon.com/AmazonRDS/latest/AuroraUserGuide/USER_PerfInsights.API.html

aws_pi_reports/time.py

@@ -0,0 +1,18 @@
+import re
+from datetime import datetime, timedelta
+from typing import Dict, Tuple, Union
+
+duration_regex_str = "^((?P<weeks>[-+\\d]+?)w|(?P<days>[-+\\d]+?)d|(?P<hours>[-+\\d]+?)h|(?P<minutes>[-+\\d]+?)m)$"
+duration_regex = re.compile(r"{duration_regex_str}")
+
+
+def parse_time(time: datetime, duration: str) -> Tuple[datetime, datetime]:
+    parts: Union[re.Match[str], None] = duration_regex.match(duration)
+    if not parts:
+        raise SyntaxError(f"Invalid duration string: {duration}. Expected format: {duration_regex_str}")
+    groups = parts.groupdict()
+    time_params: Dict[str, int] = {}
+    name = next(iter(groups))
+    value: int = int(groups[name])
+    time_params[name] = abs(value)
+    return (time, time + timedelta(**time_params)) if value > 0 else (time - timedelta(**time_params), time)

pyproject.toml

@@ -31,8 +31,9 @@ aws-pi-reports = "aws_pi_reports.main:app"
 [tool.poetry.dependencies]
 python = ">= 3.8, < 3.11"
 boto3 = "~=1.28.40"
+boto3-stubs = {extras = ["essential","pi","rds"], version = "~=1.28.40"}
 datapane = "~=0.16.7"
-typer = "~=0.9.0"
+typer = {extras = ["all"], version = "~=0.9.0"}
 # dynaconf= "^3.1.9"
 # pydantic = "^1.9.2"
 # httpx= "^0.23.0"
@@ -68,13 +69,13 @@ pipdeptree=">=2.13.0"
 
 
 [tool.black]
-line-length = 100
+line-length = 160
 target_version = ['py310']
 include = '\.py$'
 
 [tool.isort]
 profile = "black"
-line_length=100                # corresponds to -w  flag
+line_length=160                # corresponds to -w  flag
 multi_line_output=3            # corresponds to -m  flag
 include_trailing_comma=true    # corresponds to -tc flag
 skip_glob = '^((?!py$).)*$'    # isort all Python files
@@ -83,14 +84,17 @@ float_to_top=true
 [tool.mypy]
 python_version = "3.10"
 python_executable=".venv/bin/python"
+show_error_codes = true
+warn_return_any = true
+warn_unused_ignores = true
 follow_imports = "silent"
 files = ["aws_pi_reports","tests"]
 exclude = []
 disallow_any_generics = true
 disallow_untyped_defs = true
 
 [tool.pylint]
-max-line-length = 100
+max-line-length = 160
 
 [tool.ruff] # https://beta.ruff.rs/docs/configuration/#using-pyprojecttoml
 # Enable flake8-bugbear (`B`) rules.
@@ -111,7 +115,7 @@ unfixable = ["B"]
 
 [tool.pyright]
 include = ["aws_pi_reports"]
-exclude = ["tests","docs", "docker","aws_pi_reports/dynaconf"]
+exclude = ["docs", "docker","aws_pi_reports/dynaconf"] #"tests",
 defineConstant = { DEBUG = true }
 venvPath =  "."
 venv = ".venv"