flip_api.utils.s3_client
========================

.. py:module:: flip_api.utils.s3_client


Attributes
----------

.. autoapisummary::

   flip_api.utils.s3_client._MULTIPART_OVERHEAD_BUFFER_BYTES
   flip_api.utils.s3_client.MAX_PUT_PRESIGNED_URL_TTL_SECONDS


Classes
-------

.. autoapisummary::

   flip_api.utils.s3_client.S3Client


Functions
---------

.. autoapisummary::

   flip_api.utils.s3_client.parse_s3_path
   flip_api.utils.s3_client.hash_s3_key


Module Contents
---------------

.. py:data:: _MULTIPART_OVERHEAD_BUFFER_BYTES
   :value: 16384


.. py:data:: MAX_PUT_PRESIGNED_URL_TTL_SECONDS
   :value: 600


.. py:function:: parse_s3_path(s3_path: str) -> tuple[str, str]

   Parse an S3 path into bucket and key components.

   :param s3_path: Full S3 path (e.g., s3://bucket-name/key)

   :returns: Tuple containing bucket name and key, e.g. ("bucket-name", "key")


.. py:function:: hash_s3_key(key: str) -> str

   SHA-256 prefix of an S3 key, suitable for log correlation without leaking the key itself.


.. py:class:: S3Client

   S3 client wrapper for S3 operations.


   .. py:attribute:: client


   .. py:method:: get_presigned_url(s3_path: str, expiration: int = 3600) -> str

      Generate a pre-signed URL for downloading a file from S3.

      :param s3_path: Full S3 path (e.g., s3://bucket-name/key)
      :param expiration: URL expiration time in seconds (default: 1 hour)

      :returns: Pre-signed URL string
      :rtype: str

      :raises Exception: If URL generation fails



   .. py:method:: get_put_presigned_post(s3_path: str, max_bytes: int, content_type: str | None = None, expiration: int = MAX_PUT_PRESIGNED_URL_TTL_SECONDS) -> dict[str, Any]

      Generate a pre-signed POST policy for uploading a file to S3 with
      explicit size and (optional) content-type constraints baked in.

      S3 enforces the policy at the edge: multipart/form-data POSTs whose
      body exceeds ``max_bytes`` or whose Content-Type form field doesn't
      match the policy are rejected before any bytes land in the bucket.
      The single-PUT URL produced by ``generate_presigned_url("put_object",
      ...)`` carries no such constraints, which is the whole point of
      using POST here instead.

      :param s3_path: Full S3 path (e.g., ``s3://bucket-name/key``).
      :type s3_path: str
      :param max_bytes: Hard cap on the **file** size in bytes. The
                        condition actually sent to S3 is
                        ``max_bytes + _MULTIPART_OVERHEAD_BUFFER_BYTES`` because S3
                        measures the whole encoded request body, not just the file
                        part — see the module-level comment for why.
      :type max_bytes: int
      :param content_type: If provided, the policy locks
                           Content-Type to this exact value. If ``None``, any
                           Content-Type is accepted but the size cap still applies.
      :type content_type: str | None
      :param expiration: URL/policy expiration (seconds). Values above
                         ``MAX_PUT_PRESIGNED_URL_TTL_SECONDS`` are silently clamped to
                         the ceiling — a warning is logged so an over-limit caller
                         leaves an audit trail. Silent clamping is deliberate: the
                         ceiling is a hard security policy, never an error condition.
      :type expiration: int

      :returns: ``{"url": ..., "fields": {...}}`` — pass through
                to the client as multipart/form-data POST.
      :rtype: dict[str, Any]

      :raises Exception: If policy generation fails.



   .. py:method:: delete_object(s3_path: str) -> None

      Delete an object from S3 bucket.

      :param s3_path: Full S3 path (e.g., s3://bucket-name/key)

      :raises Exception: If deletion fails



   .. py:method:: delete_objects(s3_paths: list[str]) -> dict[str, Any]

      Delete multiple objects from one or more S3 buckets in grouped batch requests.

      :param s3_paths: List of full S3 paths (e.g., s3://bucket-name/key)

      :returns: Dictionary containing deletion results per bucket.
      :rtype: dict[str, Any]

      :raises Exception: If batch deletion fails for any bucket



   .. py:method:: get_object(s3_path: str) -> dict[str, Any]

      Get object from S3 bucket.

      :param s3_path: Full S3 path (e.g., s3://bucket-name/key)

      :returns: Response containing object data.
      :rtype: dict[str, Any]

      :raises EndpointConnectionError: If connection to the S3 endpoint fails.



   .. py:method:: head_object(s3_path: str) -> dict[str, Any]

      Get object metadata from S3.

      :param s3_path: Full S3 path (e.g., s3://bucket-name/key)

      :returns: Metadata of the object.
      :rtype: dict[str, Any]

      :raises Exception: If getting object metadata fails.



   .. py:method:: object_exists(s3_path: str) -> bool

      Check if an object exists in S3.

      :param s3_path: Full S3 path (e.g., s3://bucket-name/key)

      :returns: True if the object exists, False otherwise.
      :rtype: bool

      :raises ClientError: If ``head_object`` fails with any error other than HTTP 404 (e.g.,
          permissions issues).



   .. py:method:: copy_object(source_s3_path: str, dest_s3_path: str) -> None

      Copy object from source to destination bucket.

      :param source_s3_path: Full S3 path of the source object (e.g., s3://source-bucket/key)
      :param dest_s3_path: Full S3 path of the destination object (e.g., s3://dest-bucket/key)

      :raises Exception: If copying the object fails.



   .. py:method:: list_objects(s3_path: str, delimiter: str = '') -> list[str]

      List object keys under a given S3 path (non-paginated) and return full S3 paths.

      :param s3_path: Full S3 path (e.g., s3://bucket-name/prefix/)
      :param delimiter: Character used to group keys (optional)

      :returns: //bucket/key)
      :rtype: List of full S3 paths (e.g., s3

      :raises HTTPException: If listing objects fails