from argparse import ArgumentParser, Action, ArgumentError from datetime import datetime from dateutil.relativedelta import relativedelta from pytimeparse2 import parse from ..utils.datadog import get_logs_url from .base import BaseCommand class TimePeriodAction(Action): def __call__(self, parser, namespace, values, option_string=None): try: value = parse(values, raise_exception=True, as_timedelta=True) setattr(namespace, self.dest, value) except Exception as e: raise ArgumentError(self, f"Unable to parse time string: {values}") from e class CheckEmails(BaseCommand): description = "Provide link to check all notification emails been send to specified email address" def add_arguments(self, parser: ArgumentParser) -> None: parser.add_argument("-e", "--email", help="E-mail address of the user (identity)", required=True) parser.add_argument("-tp", "--time-period", action=TimePeriodAction, default=relativedelta(days=15), help="time period to check. e.g. '1h'") def run(self, *, email: str, time_period: relativedelta) -> None: query = f"@environment:prod service:daemon-notifications-delivery @resources.email:{email}" emails_url = get_logs_url(query, extra={"spanType": "all"}, start=datetime.now() - time_period) print(f"\nLink to check sent emails for user: {emails_url}\n")