Use ruff to lint

Author: RealOrangeOne

.github/workflows/ci.yml

@@ -5,7 +5,7 @@ on:
       - master
   pull_request:
 jobs:
-  test_python:
+  build:
     runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@v4
@@ -14,3 +14,13 @@ jobs:
           python-version: "3.13"
       - run: pip install -e . wheel build
       - run: python -m build
+  lint:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/setup-python@v5
+        with:
+          python-version: "3.13"
+      - run: pip install ruff
+      - run: ruff format --check .
+      - run: ruff check .

buckup/bucket_creator.py

@@ -2,17 +2,19 @@
 import time
 
 import boto3
-from botocore.exceptions import (
-    ClientError, NoCredentialsError, ParamValidationError
-)
+from botocore.exceptions import ClientError, NoCredentialsError, ParamValidationError
 
 from .exceptions import (
-    BucketNameAlreadyInUse, CannotGetCurrentUser, CannotListAccountAliases,
-    CredentialsNotFound, InvalidBucketName, InvalidUserName, UserNameTaken
+    BucketNameAlreadyInUse,
+    CannotGetCurrentUser,
+    CannotListAccountAliases,
+    CredentialsNotFound,
+    InvalidBucketName,
+    InvalidUserName,
+    UserNameTaken,
 )
 
-
-POLICY_NAME_FORMAT = '{bucket_name}-owner-policy'
+POLICY_NAME_FORMAT = "{bucket_name}-owner-policy"
 
 REQUIRE_HTTPS_CONDITION = {
     "Bool": {
@@ -21,48 +23,51 @@
     },
     "NumericGreaterThanEquals": {
         # Require TLS >= 1.2
-        "s3:TlsVersion": [
-            "1.2"
-        ]
-    }
+        "s3:TlsVersion": ["1.2"]
+    },
 }
 
 
 class BucketCreator:
     def __init__(self, profile_name=None, region_name=None):
-        self.session = boto3.session.Session(profile_name=profile_name,
-                                             region_name=region_name)
-        self.s3 = self.session.resource('s3')
-        self.s3_client = self.session.client('s3')
-        self.iam = self.session.resource('iam')
+        self.session = boto3.session.Session(
+            profile_name=profile_name, region_name=region_name
+        )
+        self.s3 = self.session.resource("s3")
+        self.s3_client = self.session.client("s3")
+        self.iam = self.session.resource("iam")
 
     def commit(self, data):
-        bucket = self.create_bucket(data['bucket_name'], data['region'])
-        user = self.create_user(bucket, data['user_name'])
+        bucket = self.create_bucket(data["bucket_name"], data["region"])
+        user = self.create_user(bucket, data["user_name"])
         self.set_bucket_policy(
             bucket,
             user,
             allow_public_acls=data["allow_public_acls"],
-            public_get_object_paths=data.get('public_get_object_paths')
+            public_get_object_paths=data.get("public_get_object_paths"),
         )
-        if data.get('cors_origins'):
-            self.set_cors(bucket, data['cors_origins'])
-        if data.get('enable_versioning'):
+        if data.get("cors_origins"):
+            self.set_cors(bucket, data["cors_origins"])
+        if data.get("enable_versioning"):
             self.enable_versioning(bucket)
 
-    def get_bucket_policy_statement_for_get_object(self, bucket, public_get_object_paths):
+    def get_bucket_policy_statement_for_get_object(
+        self, bucket, public_get_object_paths
+    ):
         """
         Create policy statement to enable the public to perform s3:getObject
         on specified paths.
         """
         if public_get_object_paths:
+
             def format_path(path):
-                if path.startswith('/'):
+                if path.startswith("/"):
                     path = path[1:]
                 return "arn:aws:s3:::{bucket_name}/{path}".format(
                     bucket_name=bucket.name,
                     path=path,
                 )
+
             paths_resources = []
             for path in public_get_object_paths:
                 paths_resources.append(format_path(path))
@@ -73,7 +78,7 @@ def format_path(path):
                 "Action": ["s3:GetObject"],
                 "Resource": paths_resources,
                 # Require HTTPS for public requests
-                "Condition": REQUIRE_HTTPS_CONDITION
+                "Condition": REQUIRE_HTTPS_CONDITION,
             }
 
     def get_bucket_policy_statements_for_user_access(self, bucket, user):
@@ -82,38 +87,32 @@ def get_bucket_policy_statements_for_user_access(self, bucket, user):
         yield {
             "Sid": "AllowUserManageBucket",
             "Effect": "Allow",
-            "Principal": {
-                "AWS": user.arn
-            },
+            "Principal": {"AWS": user.arn},
             "Action": [
                 "s3:ListBucket",
                 "s3:GetBucketLocation",
                 "s3:ListBucketMultipartUploads",
-                "s3:ListBucketVersions"
+                "s3:ListBucketVersions",
             ],
-            "Resource": "arn:aws:s3:::{bucket_name}".format(
-                bucket_name=bucket.name
-            ),
+            "Resource": "arn:aws:s3:::{bucket_name}".format(bucket_name=bucket.name),
             # Require HTTPS for API
-            "Condition": REQUIRE_HTTPS_CONDITION
+            "Condition": REQUIRE_HTTPS_CONDITION,
         }
         # Create policy statement giving the created user full access over the
         # objects.
         yield {
             "Sid": "AllowUserManageBucketObjects",
             "Effect": "Allow",
-            "Principal": {
-                "AWS": user.arn
-            },
+            "Principal": {"AWS": user.arn},
             "Action": "s3:*",
-            "Resource": "arn:aws:s3:::{bucket_name}/*".format(
-                bucket_name=bucket.name
-            ),
+            "Resource": "arn:aws:s3:::{bucket_name}/*".format(bucket_name=bucket.name),
             # Require HTTPS for API
-            "Condition": REQUIRE_HTTPS_CONDITION
+            "Condition": REQUIRE_HTTPS_CONDITION,
         }
 
-    def set_bucket_policy(self, bucket, user, allow_public_acls, public_get_object_paths=None):
+    def set_bucket_policy(
+        self, bucket, user, allow_public_acls, public_get_object_paths=None
+    ):
         policy_statement = []
         public_access = bool(public_get_object_paths)
 
@@ -124,38 +123,42 @@ def set_bucket_policy(self, bucket, user, allow_public_acls, public_get_object_p
                 "BlockPublicAcls": not allow_public_acls,
                 "IgnorePublicAcls": not allow_public_acls,
                 "BlockPublicPolicy": not public_access,
-                "RestrictPublicBuckets": not public_access
-            }
+                "RestrictPublicBuckets": not public_access,
+            },
         )
         if public_access or allow_public_acls:
-            print('Configured public access to bucket.')
+            print("Configured public access to bucket.")
 
         if public_access:
             policy_statement.append(
                 self.get_bucket_policy_statement_for_get_object(
                     bucket, public_get_object_paths
                 )
             )
-        policy_statement.extend(list(
-            self.get_bucket_policy_statements_for_user_access(bucket, user)
-        ))
-        policy = json.dumps({
-            "Version": "2012-10-17",
-            "Statement": policy_statement,
-        })
+        policy_statement.extend(
+            list(self.get_bucket_policy_statements_for_user_access(bucket, user))
+        )
+        policy = json.dumps(
+            {
+                "Version": "2012-10-17",
+                "Statement": policy_statement,
+            }
+        )
         while True:
             try:
                 bucket.Policy().put(Policy=policy)
             except ClientError as e:
-                if e.response['Error']['Code'] == 'MalformedPolicy':
-                    print('Waiting for the user to be available to be '
-                          'attached to the policy (wait 5s).')
+                if e.response["Error"]["Code"] == "MalformedPolicy":
+                    print(
+                        "Waiting for the user to be available to be "
+                        "attached to the policy (wait 5s)."
+                    )
                     time.sleep(5)
                     continue
                 raise e
             else:
                 break
-        print('Bucket policy set.')
+        print("Bucket policy set.")
 
     def create_bucket(self, name, region):
         """
@@ -164,23 +167,25 @@ def create_bucket(self, name, region):
         create_bucket_kwargs = {}
         create_bucket_config = {}
         # us-east-1 does not work with location specified.
-        if region != 'us-east-1':
-            create_bucket_config['LocationConstraint'] = region
+        if region != "us-east-1":
+            create_bucket_config["LocationConstraint"] = region
         if create_bucket_config:
-            create_bucket_kwargs['CreateBucketConfiguration'] = (
-                create_bucket_config
-            )
+            create_bucket_kwargs["CreateBucketConfiguration"] = create_bucket_config
         bucket = self.s3.Bucket(name)
         response = bucket.create(**create_bucket_kwargs)
-        msg = 'Created bucket "{bucket_name}" at "{bucket_location}" in ' \
-              'region "{region}".'
-        print(msg.format(
-            bucket_name=name,
-            bucket_location=response['Location'],
-            region=region,
-        ))
+        msg = (
+            'Created bucket "{bucket_name}" at "{bucket_location}" in '
+            'region "{region}".'
+        )
+        print(
+            msg.format(
+                bucket_name=name,
+                bucket_location=response["Location"],
+                region=region,
+            )
+        )
         print()
-        print('\tAWS_STORAGE_BUCKET_NAME', name)
+        print("\tAWS_STORAGE_BUCKET_NAME", name)
         print()
         bucket.wait_until_exists()
         return bucket
@@ -191,22 +196,22 @@ def enable_versioning(self, bucket):
 
     def create_user(self, bucket, user_name):
         user = self.iam.User(user_name).create()
-        self.iam.meta.client.get_waiter('user_exists').wait(UserName=user_name)
+        self.iam.meta.client.get_waiter("user_exists").wait(UserName=user_name)
         user.load()
-        print('Created IAM user "{user_name}".'.format(
-            user_name=user.arn
-        ))
+        print('Created IAM user "{user_name}".'.format(user_name=user.arn))
         self.create_user_access_key_pair(user)
         return user
 
     def create_user_access_key_pair(self, user):
         access_key_pair = user.create_access_key_pair()
-        print('Created access key pair for user "{user}".'.format(
-            user=user.arn,
-        ))
+        print(
+            'Created access key pair for user "{user}".'.format(
+                user=user.arn,
+            )
+        )
         print()
-        print('\tAWS_ACCESS_KEY_ID', access_key_pair.access_key_id)
-        print('\tAWS_SECRET_ACCESS_KEY', access_key_pair.secret_access_key)
+        print("\tAWS_ACCESS_KEY_ID", access_key_pair.access_key_id)
+        print("\tAWS_SECRET_ACCESS_KEY", access_key_pair.secret_access_key)
         print()
         return access_key_pair
 
@@ -216,32 +221,32 @@ def set_cors(self, bucket, origins):
             # not empty.
             next(iter(origins))
         except StopIteration:
-            raise ValueError("'origins' cannot be empty.")
+            raise ValueError("'origins' cannot be empty.") from None
         config = {
-            'CORSRules': [
+            "CORSRules": [
                 {
-                    'AllowedMethods': ['GET'],
-                    'AllowedOrigins': origins,
-                    'MaxAgeSeconds': 3000,
-                    'AllowedHeaders': ['Authorization'],
+                    "AllowedMethods": ["GET"],
+                    "AllowedOrigins": origins,
+                    "MaxAgeSeconds": 3000,
+                    "AllowedHeaders": ["Authorization"],
                 }
             ]
         }
-        msg = "Set CORS for domains {domains} to bucket \"{bucket_name}\"."
-        print(msg.format(domains=', '.join(origins), bucket_name=bucket.name))
+        msg = 'Set CORS for domains {domains} to bucket "{bucket_name}".'
+        print(msg.format(domains=", ".join(origins), bucket_name=bucket.name))
         bucket.Cors().put(CORSConfiguration=config)
 
     def validate_bucket_name(self, bucket_name):
         try:
             self.s3.meta.client.head_bucket(Bucket=bucket_name)
         except ClientError as e:
             # Bucket does not exist, proceed with creation.
-            if e.response['Error']['Code'] == '404':
+            if e.response["Error"]["Code"] == "404":
                 return
             # No access to the bucket means that it already exists but we
             # cannot run head request on it.
-            elif e.response['Error']['Code'] == '403':
-                raise BucketNameAlreadyInUse
+            elif e.response["Error"]["Code"] == "403":
+                raise BucketNameAlreadyInUse() from None
             else:
                 raise e
         except ParamValidationError as e:
@@ -253,9 +258,9 @@ def validate_user_name(self, user_name):
         try:
             self.iam.User(user_name).load()
         except ClientError as e:
-            if e.response['Error']['Code'] == 'ValidationError':
+            if e.response["Error"]["Code"] == "ValidationError":
                 raise InvalidUserName(str(e)) from e
-            if not e.response['Error']['Code'] == 'EntityAlreadyExists':
+            if not e.response["Error"]["Code"] == "EntityAlreadyExists":
                 return
             raise e
         else:
@@ -269,7 +274,7 @@ def get_current_user(self):
         except NoCredentialsError as e:
             raise CredentialsNotFound from e
         except ClientError as e:
-            if e.response['Error']['Code'] == 'AccessDenied':
+            if e.response["Error"]["Code"] == "AccessDenied":
                 raise CannotGetCurrentUser from e
             raise e
 
@@ -279,10 +284,10 @@ def get_current_account_alias(self):
         except NoCredentialsError as e:
             raise CredentialsNotFound from e
         except ClientError as e:
-            if e.response['Error']['Code'] == 'AccessDenied':
+            if e.response["Error"]["Code"] == "AccessDenied":
                 raise CannotListAccountAliases from e
             raise e
         try:
-            return response['AccountAliases'][0]
+            return response["AccountAliases"][0]
         except IndexError:
             return