[MIG] attachment_s3: Migration to 18.0

attachment_s3/__manifest__.py

@@ -5,7 +5,7 @@
 {
     "name": "Attachments on S3 storage",
     "summary": "Store assets and attachments on a S3 compatible object storage",
-    "version": "15.0.1.0.0",
+    "version": "18.0.1.0.0",
     "author": "Camptocamp,Odoo Community Association (OCA)",
     "license": "AGPL-3",
     "category": "Knowledge Management",
@@ -14,6 +14,5 @@
         "python": ["boto3"],
     },
     "website": "https://github.com/camptocamp/odoo-cloud-platform",
-    "data": [],
+    "installable": True,
-    "installable": False,
 }

attachment_s3/models/ir_attachment.py

@@ -6,8 +6,10 @@ import io
 import logging
 import os
 from urllib.parse import urlsplit
+from typing import Optional
 
-from odoo import _, api, exceptions, models
+from odoo import api, models
+from odoo.exceptions import UserError
 
 from ..s3uri import S3Uri
 
@@ -30,7 +32,7 @@ class IrAttachment(models.Model):
         return ["s3"] + super()._get_stores()
 
     @api.model
-    def _get_s3_bucket(self, name=None):
+    def _get_s3_bucket(self, name: Optional[str] = None):
         """Connect to S3 and return the bucket
 
         The following environment variables can be set:
@@ -44,29 +46,23 @@ class IrAttachment(models.Model):
         from the environment variable ``AWS_BUCKETNAME`` will be read.
 
         """
-        host = os.environ.get("AWS_HOST")
+        if not boto3:
+            raise UserError(self.env._("boto3 library is required for S3 integration."))
 
-        # Ensure host is prefixed with a scheme (use https as default)
+        host = os.getenv("AWS_HOST").strip()
+
+        # Ensure `host`` is prefixed with a scheme (use https as default)
         if host and not urlsplit(host).scheme:
-            host = "https://%s" % host
+            host = f"https://{host}"
 
-        region_name = os.environ.get("AWS_REGION")
+        region_name = os.getenv("AWS_REGION")
-        access_key = os.environ.get("AWS_ACCESS_KEY_ID")
+        access_key = os.getenv("AWS_ACCESS_KEY_ID")
-        secret_key = os.environ.get("AWS_SECRET_ACCESS_KEY")
+        secret_key = os.getenv("AWS_SECRET_ACCESS_KEY")
-        bucket_name = name or os.environ.get("AWS_BUCKETNAME")
+        bucket_name = (name or os.getenv("AWS_BUCKETNAME", "")).format(db=self.env.cr.dbname)
-        # replaces {db} by the database name to handle multi-tenancy
-        bucket_name = bucket_name.format(db=self.env.cr.dbname)
 
-        params = {
+        if not all([access_key, secret_key, bucket_name]):
-            "aws_access_key_id": access_key,
+            msg = self.env._(
-            "aws_secret_access_key": secret_key,
+                "Missing AWS credentials."
-        }
-        if host:
-            params["endpoint_url"] = host
-        if region_name:
-            params["region_name"] = region_name
-        if not (access_key and secret_key and bucket_name):
-            msg = _(
                 "If you want to read from the %(bucket_name)s S3 bucket, the following "
                 "environment variables must be set:\n"
                 "* AWS_ACCESS_KEY_ID\n"
@@ -77,101 +73,107 @@ class IrAttachment(models.Model):
                 "Optionally, the S3 host can be changed with:\n"
                 "* AWS_HOST\n"
             ).format(bucket_name=bucket_name)
+            raise UserError(msg)
 
-            raise exceptions.UserError(msg)
+        s3_params = {
-        # try:
+            "aws_access_key_id": access_key,
-        s3 = boto3.resource("s3", **params)
+            "aws_secret_access_key": secret_key,
+        }
+        if host:
+            s3_params["endpoint_url"] = host
+        if region_name:
+            s3_params["region_name"] = region_name
+
+        s3 = boto3.resource("s3", **s3_params)
         bucket = s3.Bucket(bucket_name)
-        exists = True
+
         try:
             s3.meta.client.head_bucket(Bucket=bucket_name)
         except ClientError as e:
             # If a client error is thrown, then check that it was a 404 error.
             # If it was a 404 error, then the bucket does not exist.
-            error_code = e.response["Error"]["Code"]
+            if e.response.get("Error", {}).get("Code") == "404":
-            if error_code == "404":
+                _logger.warning(f"S3 bucket '{bucket_name}' does not exist.")
-                exists = False
+                return self._create_s3_bucket(s3, bucket_name, region_name)
+            raise UserError(f"Failed to connect to S3 bucket: {str(e)}") from None
         except EndpointConnectionError as error:
             # log verbose error from s3, return short message for user
-            msg = _logger.exception("Error during connection on S3")
+            _logger.exception("Error during S3 connection.")
-            raise exceptions.UserError(str(error)) from None
+            raise UserError(str(error)) from None
 
-        if not exists:
-            if not region_name:
-                bucket = s3.create_bucket(Bucket=bucket_name)
-            else:
-                bucket = s3.create_bucket(
-                    Bucket=bucket_name,
-                    CreateBucketConfiguration={"LocationConstraint": region_name},
-                )
         return bucket
 
+    def _create_s3_bucket(self, s3, bucket_name: str, region_name: Optional[str]):
+        """Create an S3 bucket if it does not exist."""
+        params = {"Bucket": bucket_name}
+        if region_name:
+            params["CreateBucketConfiguration"] = {"LocationConstraint": region_name}
+        
+        try:
+            return s3.create_bucket(**params)
+        except ClientError as e:
+            _logger.exception(f"Failed to create S3 bucket '{bucket_name}'")
+            raise UserError(f"Bucket creation failed: {str(e)}") from None
+
     @api.model
-    def _store_file_read(self, fname):
+    def _store_file_read(self, fname: str):
         if fname.startswith("s3://"):
             s3uri = S3Uri(fname)
             try:
-                bucket = self._get_s3_bucket(name=s3uri.bucket())
+                bucket = self._get_s3_bucket(name=s3uri.bucket)
-            except exceptions.UserError:
+            except UserError:
-                _logger.exception(
+                _logger.exception(f"Error reading attachment '{fname}' from S3.")
-                    "error reading attachment '%s' from object storage", fname
-                )
                 return ""
-            try:
+
             key = s3uri.item()
+            try:
                 bucket.meta.client.head_object(Bucket=bucket.name, Key=key)
                 with io.BytesIO() as res:
                     bucket.download_fileobj(key, res)
                     res.seek(0)
-                    read = res.read()
+                    return res.read()
             except ClientError:
-                read = ""
+                _logger.info(f"Attachment '{fname}' missing on S3.")
-                _logger.info("attachment '%s' missing on object storage", fname)
+                return ""
-            return read
-        else:
         return super()._store_file_read(fname)
 
     @api.model
-    def _store_file_write(self, key, bin_data):
+    def _store_file_write(self, key: str, bin_data: bytes) -> str:
         location = self.env.context.get("storage_location") or self._storage()
         if location == "s3":
             bucket = self._get_s3_bucket()
             obj = bucket.Object(key=key)
-            with io.BytesIO() as file:
+            filename = f"s3://{bucket.name}/{key}"
-                file.write(bin_data)
+
-                file.seek(0)
-                filename = "s3://%s/%s" % (bucket.name, key)
             try:
+                with io.BytesIO(bin_data) as file:
                     obj.upload_fileobj(file)
             except ClientError as error:
                 # log verbose error from s3, return short message for user
-                    _logger.exception("Error during storage of the file %s" % filename)
+                _logger.exception(f"Error storing file {filename} on S3.")
-                    raise exceptions.UserError(
+                raise UserError(self.env._("The file could not be stored: %s") % str(error)) from None
-                        _("The file could not be stored: %s") % str(error)
+
-                    ) from None
-        else:
-            _super = super()
-            filename = _super._store_file_write(key, bin_data)
             return filename
+        return super()._store_file_write(key, bin_data)
 
     @api.model
-    def _store_file_delete(self, fname):
+    def _store_file_delete(self, fname: str):
         if fname.startswith("s3://"):
             s3uri = S3Uri(fname)
             bucket_name = s3uri.bucket()
             item_name = s3uri.item()
             # delete the file only if it is on the current configured bucket
             # otherwise, we might delete files used on a different environment
-            if bucket_name == os.environ.get("AWS_BUCKETNAME"):
+            if bucket_name == os.getenv("AWS_BUCKETNAME"):
                 bucket = self._get_s3_bucket()
                 obj = bucket.Object(key=item_name)
+
                 try:
                     bucket.meta.client.head_object(Bucket=bucket.name, Key=item_name)
                     obj.delete()
-                    _logger.info("file %s deleted on the object storage" % (fname,))
+                    _logger.info(f"File {fname} deleted from S3.")
                 except ClientError:
-                    # log verbose error from s3, return short message for
+                    # log verbose error from s3, return short message for user
-                    # user
+                    _logger.exception(f"Error deleting file {fname} from S3.")
-                    _logger.exception("Error during deletion of the file %s" % fname)
         else:
             return super()._store_file_delete(fname)

attachment_s3/pyproject.toml

@@ -0,0 +1,3 @@
+[build-system]
+requires = ["whool"]
+build-backend = "whool.buildapi"

attachment_s3/s3uri.py

@@ -2,19 +2,23 @@
 # License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl.html)
 
 import re
-
+from typing import Optional
 
 class S3Uri:
-    _url_re = re.compile("^s3:///*([^/]*)/?(.*)", re.IGNORECASE | re.UNICODE)
+    _url_re = re.compile(r"^s3://+([^/]+)/?(.*)", re.IGNORECASE | re.UNICODE)
 
-    def __init__(self, uri):
+    def __init__(self, uri: str) -> None:
         match = self._url_re.match(uri)
         if not match:
-            raise ValueError("%s: is not a valid S3 URI" % (uri,))
+            raise ValueError(f"{uri}: is not a valid S3 URI")
-        self._bucket, self._item = match.groups()
         
-    def bucket(self):
+        self._bucket: str = match.group(1)
+        self._item: Optional[str] = match.group(2) if match.group(2) else None
+
+    @property
+    def bucket(self) -> str:
         return self._bucket
 
-    def item(self):
+    @property
+    def item(self) -> Optional[str]:
         return self._item