style: Reformat with ruff (#5)

cli.py

@@ -7,15 +7,17 @@
 from commands.customers import customers
 from commands.awslambda import awslambda
 
+
 @click.group()
 def cli():
     pass
 
+
 cli.add_command(cognito)
 cli.add_command(handle)
 cli.add_command(users)
 cli.add_command(customers)
 cli.add_command(awslambda)
 
 if __name__ == "__main__":
-    cli()
+    cli()

commands/awslambda.py

@@ -1,12 +1,19 @@
 import click
 from commands.services.lambda_api import LambdaService
 
+
 @click.group()
 def awslambda():
     pass
 
+
 @awslambda.command(help="Delete old versions of AWS Lambda functions.")
-@click.option('--profile', envvar='AWS_PROFILE', default='default', help='The AWS profile to use. e.g. sikt-nva-sandbox, configure your profiles in ~/.aws/config')
-@click.option('--delete', is_flag=True, default=True, help='Delete old versions.')
-def delete_old_versions(profile:str, delete:bool) -> None:
-    LambdaService(profile).delete_old_versions(delete)
+@click.option(
+    "--profile",
+    envvar="AWS_PROFILE",
+    default="default",
+    help="The AWS profile to use. e.g. sikt-nva-sandbox, configure your profiles in ~/.aws/config",
+)
+@click.option("--delete", is_flag=True, default=True, help="Delete old versions.")
+def delete_old_versions(profile: str, delete: bool) -> None:
+    LambdaService(profile).delete_old_versions(delete)

commands/cognito.py

@@ -3,14 +3,21 @@
 from commands.services.cognito_api import CognitoService
 from commands.services.aws_utils import prettify
 
+
 @click.group()
 def cognito():
     pass
 
+
 @cognito.command(help="Search users by user attribute values")
-@click.option('--profile', envvar='AWS_PROFILE', default='default', help='The AWS profile to use. e.g. sikt-nva-sandbox, configure your profiles in ~/.aws/config')
-@click.argument('search_term', required=True, nargs=-1)
-def search(profile:str, search_term:str)  -> None:
-    search_term = ' '.join(search_term)
+@click.option(
+    "--profile",
+    envvar="AWS_PROFILE",
+    default="default",
+    help="The AWS profile to use. e.g. sikt-nva-sandbox, configure your profiles in ~/.aws/config",
+)
+@click.argument("search_term", required=True, nargs=-1)
+def search(profile: str, search_term: str) -> None:
+    search_term = " ".join(search_term)
     result = CognitoService(profile).search(search_term)
-    click.echo(prettify(result))
+    click.echo(prettify(result))

commands/customers.py

@@ -1,20 +1,38 @@
 import click
 
-from commands.services.customers_api import list_missing_customers, list_duplicate_customers
+from commands.services.customers_api import (
+    list_missing_customers,
+    list_duplicate_customers,
+)
 from commands.services.aws_utils import prettify
 
+
 @click.group()
 def customers():
     pass
 
-@customers.command(help="Search customer references from users that does not exsist in the customer table")
-@click.option('--profile', envvar='AWS_PROFILE', default='default', help='The AWS profile to use. e.g. sikt-nva-sandbox, configure your profiles in ~/.aws/config')
+
+@customers.command(
+    help="Search customer references from users that does not exsist in the customer table"
+)
+@click.option(
+    "--profile",
+    envvar="AWS_PROFILE",
+    default="default",
+    help="The AWS profile to use. e.g. sikt-nva-sandbox, configure your profiles in ~/.aws/config",
+)
 def list_missing(profile) -> None:
     result = list_missing_customers(profile)
     click.echo(prettify(result))
 
+
 @customers.command(help="Search dubplicate customer references (same cristin id)")
-@click.option('--profile', envvar='AWS_PROFILE', default='default', help='The AWS profile to use. e.g. sikt-nva-sandbox, configure your profiles in ~/.aws/config')
-def list_duplicate(profile:str) -> None:
+@click.option(
+    "--profile",
+    envvar="AWS_PROFILE",
+    default="default",
+    help="The AWS profile to use. e.g. sikt-nva-sandbox, configure your profiles in ~/.aws/config",
+)
+def list_duplicate(profile: str) -> None:
     result = list_duplicate_customers(profile)
-    click.echo(prettify(result))
+    click.echo(prettify(result))

commands/handle.py

@@ -4,25 +4,51 @@
 import shutil
 from boto3.dynamodb.conditions import Key
 from commands.services.handle_task_writer import HandleTaskWriterService
-from commands.services.handle_task_executor import  HandleTaskExecutorService
+from commands.services.handle_task_executor import HandleTaskExecutorService
 from commands.services.dynamodb_export import DynamodbExport, get_account_alias
 
+
 @click.group()
 def handle():
     pass
 
+
 @handle.command()
-@click.option('--profile', envvar='AWS_PROFILE', default='default', help='The AWS profile to use. e.g. sikt-nva-sandbox, configure your profiles in ~/.aws/config')
-@click.option('-c', '--customer', required=True, help='Customer UUID. e.g. bb3d0c0c-5065-4623-9b98-5810983c2478')
-@click.option('-r', '--resource-owner', required=True, help='Resource owner ID. e.g. ntnu@194.0.0.0')
-@click.option('-o', '--output-folder', required=False, help='Output folder path. e.g. sikt-nva-sandbox-resources-ntnu@194.0.0.0-handle-tasks')
-def prepare(profile:str, customer:str, resource_owner:str, output_folder:str) -> None:
-    table_pattern = '^nva-resources-master-pipelines-NvaPublicationApiPipeline-.*-nva-publication-api$'
-    condition = Key('PK0').eq(f'Resource:{customer}:{resource_owner}')
+@click.option(
+    "--profile",
+    envvar="AWS_PROFILE",
+    default="default",
+    help="The AWS profile to use. e.g. sikt-nva-sandbox, configure your profiles in ~/.aws/config",
+)
+@click.option(
+    "-c",
+    "--customer",
+    required=True,
+    help="Customer UUID. e.g. bb3d0c0c-5065-4623-9b98-5810983c2478",
+)
+@click.option(
+    "-r",
+    "--resource-owner",
+    required=True,
+    help="Resource owner ID. e.g. ntnu@194.0.0.0",
+)
+@click.option(
+    "-o",
+    "--output-folder",
+    required=False,
+    help="Output folder path. e.g. sikt-nva-sandbox-resources-ntnu@194.0.0.0-handle-tasks",
+)
+def prepare(
+    profile: str, customer: str, resource_owner: str, output_folder: str
+) -> None:
+    table_pattern = "^nva-resources-master-pipelines-NvaPublicationApiPipeline-.*-nva-publication-api$"
+    condition = Key("PK0").eq(f"Resource:{customer}:{resource_owner}")
     batch_size = 700
 
     if not output_folder:
-        output_folder = f'{get_account_alias(profile)}-resources-{resource_owner}-handle-tasks'
+        output_folder = (
+            f"{get_account_alias(profile)}-resources-{resource_owner}-handle-tasks"
+        )
 
     # Create output folder if not exists
     if not os.path.exists(output_folder):
@@ -31,13 +57,13 @@ def prepare(profile:str, customer:str, resource_owner:str, output_folder:str) ->
     action_counts = {}
 
     def process_batch(batch, batch_counter):
-        with open(f'{output_folder}/batch_{batch_counter}.jsonl', 'w') as outfile:
+        with open(f"{output_folder}/batch_{batch_counter}.jsonl", "w") as outfile:
             for data in batch:
                 task = HandleTaskWriterService().process_item(data)
-                action = task.get('action')
+                action = task.get("action")
                 action_counts[action] = action_counts.get(action, 0) + 1
                 json.dump(task, outfile)
-                outfile.write('\n')
+                outfile.write("\n")
 
     DynamodbExport(profile, table_pattern, condition, batch_size).process(process_batch)
 
@@ -47,19 +73,29 @@ def process_batch(batch, batch_counter):
 
 
 @handle.command()
-@click.option('--profile', envvar='AWS_PROFILE', default='default', help='The AWS profile to use. e.g. sikt-nva-sandbox, configure your profiles in ~/.aws/config')
-@click.option('-i', '--input-folder', required=True, help='Input folder path. e.g. sikt-nva-sandbox-resources-ntnu@194.0.0.0-handle-tasks')
-def execute(profile:str, input_folder:str) -> None:
-    complete_folder = os.path.join(input_folder, 'complete')
+@click.option(
+    "--profile",
+    envvar="AWS_PROFILE",
+    default="default",
+    help="The AWS profile to use. e.g. sikt-nva-sandbox, configure your profiles in ~/.aws/config",
+)
+@click.option(
+    "-i",
+    "--input-folder",
+    required=True,
+    help="Input folder path. e.g. sikt-nva-sandbox-resources-ntnu@194.0.0.0-handle-tasks",
+)
+def execute(profile: str, input_folder: str) -> None:
+    complete_folder = os.path.join(input_folder, "complete")
     os.makedirs(complete_folder, exist_ok=True)
 
     for batch_file in os.listdir(input_folder):
         file_path = os.path.join(input_folder, batch_file)
         if os.path.isfile(file_path):
-            with open(file_path, 'r') as infile:
+            with open(file_path, "r") as infile:
                 batch = [json.loads(line) for line in infile]
                 HandleTaskExecutorService(profile).execute(batch)
-            
+
             # Move the file to the 'complete' folder after processing
             new_file_path = os.path.join(complete_folder, batch_file)
             shutil.move(file_path, new_file_path)

commands/services/aws_utils.py

@@ -2,18 +2,20 @@
 import boto3
 import json
 
-def get_account_alias(profile:str=None) -> str:
+
+def get_account_alias(profile: str = None) -> str:
     # Create a default Boto3 session
     session = boto3.Session(profile_name=profile) if profile else boto3.Session()
 
     # Create an IAM client
-    iam = session.client('iam')
+    iam = session.client("iam")
 
     # Get the account alias
-    account_aliases = iam.list_account_aliases()['AccountAliases']
+    account_aliases = iam.list_account_aliases()["AccountAliases"]
 
     # Return the first account alias or None if the list is empty
     return account_aliases[0] if account_aliases else None
 
+
 def prettify(object) -> str:
-    return json.dumps(object, indent=2, sort_keys=True, default=str, ensure_ascii=False)
+    return json.dumps(object, indent=2, sort_keys=True, default=str, ensure_ascii=False)

commands/services/cognito_api.py

@@ -1,5 +1,6 @@
 import boto3
 
+
 class CognitoService:
     def __init__(self, profile):
         self.profile = profile
@@ -11,46 +12,55 @@ def search(self, search_term):
         return self._lookup_users_by_attribute_value(search_term, users)
 
     def _get_user_pool_id(self):
-        session = boto3.Session(profile_name=self.profile) if self.profile else boto3.Session()
-        client = session.client('ssm')
+        session = (
+            boto3.Session(profile_name=self.profile)
+            if self.profile
+            else boto3.Session()
+        )
+        client = session.client("ssm")
 
-        parameter_name = 'CognitoUserPoolId'
+        parameter_name = "CognitoUserPoolId"
 
-        response = client.get_parameter(
-            Name=parameter_name,
-            WithDecryption=True
-        )
+        response = client.get_parameter(Name=parameter_name, WithDecryption=True)
+
+        return response["Parameter"]["Value"]
 
-        return response['Parameter']['Value']
-
     def _get_all_users(self, user_pool_id):
-        session = boto3.Session(profile_name=self.profile) if self.profile else boto3.Session()
-        cognito = session.client('cognito-idp')
-
+        session = (
+            boto3.Session(profile_name=self.profile)
+            if self.profile
+            else boto3.Session()
+        )
+        cognito = session.client("cognito-idp")
+
         pagination_token = None
         users = []
 
         while True:
             if pagination_token:
-                response = cognito.list_users(UserPoolId=user_pool_id, PaginationToken=pagination_token)
+                response = cognito.list_users(
+                    UserPoolId=user_pool_id, PaginationToken=pagination_token
+                )
             else:
                 response = cognito.list_users(UserPoolId=user_pool_id)
 
-            users.extend(response['Users'])
+            users.extend(response["Users"])
 
-            pagination_token = response.get('PaginationToken')
+            pagination_token = response.get("PaginationToken")
             if not pagination_token:
                 break
 
         return users
-    
+
     def _lookup_users_by_attribute_value(self, search_term, users):
         search_words = search_term.split()
         matches = []
 
         for user in users:
-            user_attributes = ' '.join(attribute['Value'] for attribute in user['Attributes'])
+            user_attributes = " ".join(
+                attribute["Value"] for attribute in user["Attributes"]
+            )
             if all(word in user_attributes for word in search_words):
                 matches.append(user)
 
-        return matches if matches else None
+        return matches if matches else None