runner.py

"""
Copyright 2025 The Flame Authors.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
    http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
"""

import inspect
import io
import logging
import os
import tarfile
from concurrent.futures import Future, as_completed
from typing import Any, Callable, List, Optional

import cloudpickle

from flamepy.core import ObjectRef, get_object, put_object
from flamepy.core.client import get_application, open_session, register_application, unregister_application
from flamepy.core.types import (
    ApplicationAttributes,
    FlameContext,
    FlameError,
    FlameErrorCode,
    ResourceRequirement,
    SessionAttributes,
    short_name,
)
from flamepy.runner.storage import StorageBackend, create_storage_backend
from flamepy.runner.types import (
    RunnerContext,
    RunnerRequest,
    SessionContext,
)

logger = logging.getLogger(__name__)


class ObjectFuture:
    """Encapsulates a future that resolves to an ObjectRef.

    This class manages asynchronous and deferred computation results in runner services.
    The underlying future is expected to always yield an ObjectRef instance when resolved.

    Attributes:
        _future: A Future that will resolve to an ObjectRef
    """

    def __init__(self, future: Future):
        """Initialize an ObjectFuture.

        Args:
            future: A Future that resolves to an ObjectRef
        """
        self._future = future

    def ref(self) -> ObjectRef:
        """Get the ObjectRef by waiting for the future to complete.

        This method is primarily intended for internal use within the Flame SDK,
        providing direct access to the encapsulated object reference.

        Returns:
            The ObjectRef from the completed future
        """
        result = self._future.result()
        # The future returns bytes (ObjectRef encoded), decode it to ObjectRef
        if isinstance(result, bytes):
            return ObjectRef.decode(result)
        # If it's already an ObjectRef, return it as-is
        if isinstance(result, ObjectRef):
            return result
        # Otherwise, assume it's bytes and try to decode
        return ObjectRef.decode(result)

    def get(self) -> Any:
        """Retrieve the concrete object that this ObjectFuture represents.

        This method fetches the ObjectRef via the future, then uses cache.get_object
        to retrieve the actual underlying object.

        Returns:
            The deserialized object from the cache
        """
        result = self._future.result()
        # The future returns bytes (ObjectRef encoded), decode it to ObjectRef
        if isinstance(result, bytes):
            object_ref = ObjectRef.decode(result)
        elif isinstance(result, ObjectRef):
            object_ref = result
        else:
            # Otherwise, assume it's bytes and try to decode
            object_ref = ObjectRef.decode(result)
        return get_object(object_ref)

    def wait(self) -> None:
        """Wait for the future to complete without fetching the result."""
        self._future.result()


class ObjectFutureIterator:
    """Iterator wrapper over futures that yields ObjectFuture as they complete."""

    def __init__(self, futures: List[ObjectFuture]):
        self._future_map = {future._future: future for future in futures}

    def __iter__(self):
        for future in as_completed(self._future_map):
            yield self._future_map[future]


class RunnerService:
    """Encapsulates an execution object for remote invocation within Flame.

    This class creates a session with the flamepy.runner.runpy service and dynamically
    generates wrapper methods for all public methods of the execution object.
    Each wrapper submits tasks to the session and returns ObjectFuture instances.

    Attributes:
        _app: The name of the application registered in Flame
        _execution_object: The Python execution object being managed
        _session: The Flame session for task execution
    """

    def __init__(
        self,
        app: str,
        execution_object: Any,
        autoscale: Optional[bool] = None,
        warmup: int = 0,
        resreq: Optional[ResourceRequirement] = None,
    ):
        """Initialize a RunnerService.

        Args:
            app: The name of the application registered in Flame.
                 The associated service must be flamepy.runner.runpy.
            execution_object: The Python execution object to be managed and
                             exposed as a remote service. Can be a class, instance,
                             or function. If the object has a `_session_context`
                             attribute of type SessionContext, its session_id
                             will be used instead of auto-generating one.
            autoscale: For functions, builtins, and classes, whether to create
                      instances dynamically based on pending tasks. Defaults to
                      True for those service types. Object instances are always
                      fixed and cannot autoscale.
            warmup: Number of instances to pre-create at session start. When
                    autoscale=False, this sets the fixed instance count. Object
                    instances only support warmup values 0 and 1.
            resreq: Optional explicit resource requirements. When omitted, the
                    server applies cluster.resource_requirement (or a hardcoded
                    fallback when that is unset).
        """
        self._app = app
        self._execution_object = execution_object
        self._function_wrapper = None  # For callable functions

        # Extract custom session_id from SessionContext if present
        custom_session_id = None
        if hasattr(execution_object, "_session_context"):
            ctx = getattr(execution_object, "_session_context")
            if isinstance(ctx, SessionContext):
                custom_session_id = ctx.session_id
                if ctx.application_name:
                    logger.debug(f"SessionContext application_name: {ctx.application_name}")
            else:
                logger.warning(f"_session_context attribute found but is not SessionContext (got {type(ctx).__name__}), ignoring")

        # Determine session_id: use custom if provided, otherwise generate
        session_id = custom_session_id if custom_session_id else short_name(app)

        # Create a session with flamepy.runner.runpy service
        # For RL module: serialize RunnerContext with cloudpickle, put in cache to get ObjectRef,
        # then encode ObjectRef to bytes for core API
        runner_context = RunnerContext(execution_object=execution_object, autoscale=autoscale, warmup=warmup)
        # Serialize the context using cloudpickle
        serialized_ctx = cloudpickle.dumps(runner_context, protocol=cloudpickle.DEFAULT_PROTOCOL)
        # Put in cache with <app>/<session_id> key prefix
        key_prefix = f"{app}/{session_id}"
        logger.debug(f"[RunnerService] Putting RunnerContext in cache: key_prefix={key_prefix}, stateful={runner_context.stateful}, autoscale={runner_context.autoscale}")
        object_ref = put_object(key_prefix, serialized_ctx)
        logger.debug(f"[RunnerService] RunnerContext cached: key={object_ref.key}, version={object_ref.version}")
        # Encode ObjectRef to bytes for core API
        common_data_bytes = object_ref.encode()
        session_spec = SessionAttributes(
            id=session_id,
            application=app,
            common_data=common_data_bytes,
            min_instances=runner_context.min_instances,
            max_instances=runner_context.max_instances,
            batch_size=1,
            resreq=resreq,
        )
        logger.info(f"[RunnerService] Opening session: session_id={session_id}, app={app}")
        try:
            self._session = open_session(session_id=session_id, spec=session_spec)
        except Exception as e:
            logger.error(f"[RunnerService] Failed to open session: {type(e).__name__}: {e}", exc_info=True)
            raise

        logger.info(f"[RunnerService] Session opened: id={self._session.id}")

        # Generate wrapper methods for all public methods of the execution object
        self._generate_wrappers()

    def _generate_wrappers(self) -> None:
        """Generate wrapper functions for all public methods of the execution object.

        This method inspects the execution object and creates a wrapper for each
        public method (not starting with '_'). Each wrapper:
        - Converts ObjectFuture arguments to ObjectRef
        - Constructs a RunnerRequest
        - Submits a task via _session.run()
        - Returns an ObjectFuture
        """
        # Determine if execution_object is a function or has methods
        if callable(self._execution_object) and not inspect.isclass(self._execution_object):
            # It's a function, create a wrapper for direct invocation
            self._create_function_wrapper()
        else:
            # It's a class or instance, wrap all public methods
            self._create_method_wrappers()

    def _create_function_wrapper(self) -> None:
        """Create a wrapper for a callable execution object (function)."""

        def wrapper(*args, **kwargs):
            # Convert ObjectFuture arguments to ObjectRef
            converted_args = tuple(arg.ref() if isinstance(arg, ObjectFuture) else arg for arg in args)
            converted_kwargs = {key: value.ref() if isinstance(value, ObjectFuture) else value for key, value in kwargs.items()}

            # Create a RunnerRequest with method=None for direct callable invocation
            request = RunnerRequest(
                method=None,
                args=converted_args if converted_args else None,
                kwargs=converted_kwargs if converted_kwargs else None,
            )

            # For RL module: serialize RunnerRequest with cloudpickle, then call core API
            request_bytes = cloudpickle.dumps(request, protocol=cloudpickle.DEFAULT_PROTOCOL)
            # Submit task and return ObjectFuture
            future = self._session.run(request_bytes)
            return ObjectFuture(future)

        # Store the wrapper so __call__ can use it
        self._function_wrapper = wrapper
        logger.debug("Created callable wrapper for function execution object")

    def _create_method_wrappers(self) -> None:
        """Create wrappers for all public methods of a class/instance."""
        # Get all public methods (not starting with '_')
        for attr_name in dir(self._execution_object):
            if attr_name.startswith("_"):
                continue

            attr = getattr(self._execution_object, attr_name)
            if not callable(attr):
                continue

            if hasattr(type(self), attr_name) or attr_name in self.__dict__:
                logger.warning(
                    "Skipping wrapper for method '%s' because it conflicts with RunnerService",
                    attr_name,
                )
                continue

            # Create a wrapper for this method
            wrapper = self._create_method_wrapper(attr_name)
            setattr(self, attr_name, wrapper)
            logger.debug(f"Created wrapper for method '{attr_name}'")

    def _create_method_wrapper(self, method_name: str) -> Callable:
        """Create a wrapper function for a specific method.

        Args:
            method_name: The name of the method to wrap

        Returns:
            A wrapper function that submits tasks and returns ObjectFuture
        """

        def wrapper(*args, **kwargs):
            # Convert ObjectFuture arguments to ObjectRef
            converted_args = tuple(arg.ref() if isinstance(arg, ObjectFuture) else arg for arg in args)
            converted_kwargs = {key: value.ref() if isinstance(value, ObjectFuture) else value for key, value in kwargs.items()}

            # Create a RunnerRequest for this method
            request = RunnerRequest(
                method=method_name,
                args=converted_args if len(converted_args) > 0 else None,
                kwargs=converted_kwargs if converted_kwargs and len(converted_kwargs) > 0 else None,
            )

            # For RL module: serialize RunnerRequest with cloudpickle, then call core API
            request_bytes = cloudpickle.dumps(request, protocol=cloudpickle.DEFAULT_PROTOCOL)
            logger.info(f"[RunnerService] Submitting task: method={method_name}, session={self._session.id}")
            # Submit task and return ObjectFuture
            future = self._session.run(request_bytes)
            return ObjectFuture(future)

        return wrapper

    def __call__(self, *args, **kwargs) -> ObjectFuture:
        """Make RunnerService callable for function execution objects.

        This method allows calling the service directly when the execution object
        is a function (not a class or instance).

        Args:
            *args: Positional arguments to pass to the function
            **kwargs: Keyword arguments to pass to the function

        Returns:
            ObjectFuture that resolves to the function's result

        Raises:
            TypeError: If the execution object is not a callable function
        """
        if self._function_wrapper is None:
            raise TypeError(f"RunnerService for app '{self._app}' is not callable. The execution object is a class or instance, not a function. Call specific methods instead.")
        return self._function_wrapper(*args, **kwargs)

    def close(self) -> None:
        """Gracefully close the RunnerService and clean up resources.

        This closes the underlying session.
        """
        logger.debug(f"Closing RunnerService for app '{self._app}'")
        self._session.close()


class Runner:
    """Context manager for managing lifecycle and deployment of Python packages in Flame.

    This class automates the packaging, uploading, registration, and cleanup of
    Python applications within Flame. It can be used either as a context manager
    or with explicit close() call.

    Attributes:
        _name: The name of the application/package
        _services: List of RunnerService instances created within this context
        _package_path: Path to the created package file
        _app_registered: Whether the application was successfully registered
        _storage_backend: Storage backend instance for uploading/deleting packages
        _started: Whether the runner has been started
        _fail_if_exists: Whether to raise an exception if the application already exists
        _dependencies: List of pip dependencies for auto-generated pyproject.toml
        _python_version: Optional Python version to use for execution (e.g., "3.12")
    """

    def __init__(
        self,
        name: str,
        fail_if_exists: bool = False,
        dependencies: Optional[List[str]] = None,
        python_version: Optional[str] = None,
    ):
        """Initialize and start a Runner.

        Args:
            name: The name of the application/package
            fail_if_exists: If True, raise an exception if the application already exists.
                           If False (default), skip registration if the application already exists.
            dependencies: List of pip dependencies (e.g., ["numpy", "pandas>=2.0"]).
                         If provided and no pyproject.toml exists, one will be auto-generated.
            python_version: Python version to use for execution.
                           If omitted, the executor uses the latest installed Flame Python SDK.
        """
        self._name = name
        self._services: List[RunnerService] = []
        self._package_path: Optional[str] = None
        self._app_registered = False
        self._context = FlameContext()
        self._storage_backend: Optional[StorageBackend] = None
        self._started = False
        self._fail_if_exists = fail_if_exists
        self._dependencies = dependencies
        self._python_version = python_version

        logger.debug(f"Initialized Runner '{name}' (fail_if_exists={fail_if_exists}, dependencies={dependencies}, python_version={python_version})")

        self._start()

    def _start(self) -> None:
        """Internal method to start the runner and set up the application environment.

        Steps:
        1. Check if application already exists (skip packaging if reusing)
        2. Package the current working directory into a .tar.gz archive
        3. Upload the package to the storage location
        4. Retrieve the flmrun application template
        5. Register a new application with the package URL

        Raises:
            FlameError: If setup fails at any step
        """
        if self._started:
            logger.debug(f"Runner '{self._name}' already started, skipping")
            return

        logger.debug(f"Starting Runner '{self._name}'")

        # Check if application already exists first (before packaging)
        existing_app = get_application(self._name)
        if existing_app is not None:
            if self._fail_if_exists:
                raise FlameError(FlameErrorCode.ALREADY_EXISTS, f"Application '{self._name}' already exists. Set fail_if_exists=False to skip registration.")
            else:
                logger.debug(f"[Runner._start] Application '{self._name}' already exists, skipping registration")
                self._started = True
                return

        # Initialize storage backend (uses cache.endpoint if package.storage not set)
        storage_base = self._context.package.storage if self._context.package else None
        if storage_base is None and self._context.cache is None:
            raise FlameError(FlameErrorCode.INVALID_CONFIG, "Storage not configured. Please set 'cache.endpoint' or 'package.storage' in flame.yaml.")
        self._storage_backend = create_storage_backend(storage_base, app_name=self._name)
        logger.debug(f"Initialized storage backend: {type(self._storage_backend).__name__}")

        # Step 1: Package the current working directory
        self._package_path = self._create_package()
        logger.debug(f"Created package: {self._package_path}")

        # Step 2: Upload the package to storage
        storage_url = self._upload_package()
        logger.debug(f"Uploaded package to: {storage_url}")

        # Step 3: Retrieve the application template
        # Use configured template if available, otherwise default to flmrun
        template_name = self._context.runner.template

        try:
            template_app = get_application(template_name)
            logger.debug(f"Retrieved application template: {template_name}")
        except Exception as e:
            # Clean up the package file
            if self._package_path and os.path.exists(self._package_path):
                os.remove(self._package_path)
            raise FlameError(FlameErrorCode.INTERNAL, f"Failed to get application template '{template_name}': {str(e)}")

        # Register the new application
        try:
            working_directory = None
            if template_app.working_directory is not None and template_app.working_directory != "":
                working_directory = f"{template_app.working_directory}/{self._name}"

            logger.debug(f"Working directory: {working_directory}")

            environments = dict(template_app.environments) if template_app.environments else {}
            if self._python_version:
                environments["FLAME_PYTHON_VERSION"] = self._python_version

            app_attrs = ApplicationAttributes(
                image=template_app.image,
                command=template_app.command,
                description=f"Runner application: {self._name}",
                labels=template_app.labels,
                arguments=template_app.arguments,
                environments=environments,
                working_directory=working_directory,
                max_instances=template_app.max_instances,
                delay_release=template_app.delay_release,
                schema=template_app.schema,
                url=storage_url,
                installer=template_app.installer,
            )

            register_application(self._name, app_attrs)
            self._app_registered = True
            self._started = True
            logger.debug(f"Registered application '{self._name}' with working directory: {working_directory}")
        except FlameError:
            raise
        except Exception as e:
            self._cleanup_storage()
            if self._package_path and os.path.exists(self._package_path):
                os.remove(self._package_path)
            raise FlameError(FlameErrorCode.INTERNAL, f"Failed to register application: {str(e)}")

    def __enter__(self) -> "Runner":
        """Enter the context manager and set up the application environment.

        Returns:
            self for use in the with statement

        Raises:
            FlameError: If setup fails at any step
        """
        self._start()
        return self

    def __exit__(self, exc_type, exc_val, exc_tb) -> None:
        """Exit the context manager and clean up resources."""
        self.close()

    def close(self) -> None:
        """Close the Runner and clean up all resources.

        This method can be called explicitly or is automatically called when
        exiting the context manager. It performs the following cleanup:
        1. Closes all RunnerService instances (only if app was registered by this Runner)
        2. Deletes all cached objects for this application from flame-cache
        3. Unregisters the application (only if registered by this Runner instance)
        4. Deletes the package from storage (only if uploaded by this Runner)
        5. Removes the local package file (only if created by this Runner)

        Note: If the application already existed when the Runner was created
        (fail_if_exists=False), the Runner will NOT perform any cleanup.
        This allows recursive runners to reuse existing applications without
        affecting their lifecycle.
        """
        if not self._started:
            logger.debug(f"Runner '{self._name}' not started, nothing to close")
            return

        # If this Runner did not register the application, skip all cleanup
        # to allow recursive/nested runners to reuse existing apps safely
        if not self._app_registered:
            logger.debug(f"Runner '{self._name}' did not register app, skipping cleanup")
            self._started = False
            return

        logger.debug(f"Closing Runner '{self._name}'")

        for service in self._services:
            try:
                service.close()
            except Exception as e:
                logger.error(f"Error closing service: {e}", exc_info=True)

        try:
            unregister_application(self._name)
            self._app_registered = False
            logger.debug(f"Unregistered application '{self._name}'")
        except Exception as e:
            logger.error(f"Error unregistering application: {e}", exc_info=True)

        try:
            from flamepy.core.cache import ObjectKey, delete_objects

            delete_objects(ObjectKey.for_all_sessions(self._name).to_prefix())
            logger.debug(f"Deleted cached objects for '{self._name}'")
        except Exception as e:
            logger.error(f"Error deleting cached objects: {e}", exc_info=True)

        self._cleanup_storage()

        if self._package_path and os.path.exists(self._package_path):
            try:
                os.remove(self._package_path)
                logger.debug(f"Removed local package: {self._package_path}")
            except Exception as e:
                logger.error(f"Error removing local package: {e}", exc_info=True)

        self._started = False

    def service(
        self,
        execution_object: Any,
        autoscale: Optional[bool] = None,
        warmup: int = 0,
        resreq: Optional[ResourceRequirement] = None,
    ) -> RunnerService:
        """Create a RunnerService for the given execution object.

        Args:
            execution_object: A function, class, or class instance to expose as a service
            autoscale: Functions, builtins, and classes can autoscale; their default
                      is True. Object instances are fixed and cannot autoscale.
            warmup: Number of instances to pre-create at session start. When
                   autoscale=False, this sets the fixed instance count. With
                   warmup=0, fixed services create one instance and autoscaled
                   services start from zero. Object instances only support warmup
                   values 0 and 1. Default: 0.
            resreq: Optional explicit resource requirements. When omitted, the
                    server applies cluster.resource_requirement (or a hardcoded
                    fallback when that is unset).

        Returns:
            A RunnerService instance

        Raises:
            ValueError: If the requested stateful/autoscale/warmup combination is
                not supported for the execution object type.
        """
        logger.debug(f"Creating service for {type(execution_object).__name__} (autoscale={autoscale}, warmup={warmup})")

        runner_service = RunnerService(
            self._name,
            execution_object,
            autoscale=autoscale,
            warmup=warmup,
            resreq=resreq,
        )
        self._services.append(runner_service)

        logger.debug(f"Created service for execution object in Runner '{self._name}'")
        return runner_service

    def get(self, futures: List[ObjectFuture]) -> List[Any]:
        """Resolve multiple ObjectFuture values to their concrete results.

        Args:
            futures: List of ObjectFuture instances

        Returns:
            List of concrete results corresponding to each ObjectFuture
        """
        return [future.get() for future in futures]

    def ref(self, futures: List[ObjectFuture]) -> List[ObjectRef]:
        """Resolve multiple ObjectFuture values to their ObjectRef references.

        Args:
            futures: List of ObjectFuture instances

        Returns:
            List of ObjectRef instances corresponding to each ObjectFuture
        """
        return [future.ref() for future in futures]

    def wait(self, futures: List[ObjectFuture]) -> None:
        """Wait for multiple ObjectFuture values to complete.

        Args:
            futures: List of ObjectFuture instances
        """
        for future in futures:
            future.wait()

    def select(self, futures: List[ObjectFuture]) -> ObjectFutureIterator:
        """Return an iterator over futures as they complete.

        Args:
            futures: List of ObjectFuture instances

        Returns:
            ObjectFutureIterator yielding futures in completion order
        """
        return ObjectFutureIterator(futures)

    def put_object(self, obj: Any) -> ObjectRef:
        """Put an object into the cache with <app_name>/shared key prefix.

        Args:
            obj: The object to cache (will be pickled)

        Returns:
            ObjectRef pointing to the cached object
        """
        from flamepy.core.cache import ObjectKey, put_object

        object_key = ObjectKey.for_shared(self._name)
        return put_object(object_key.to_prefix(), obj)

    def _create_package(self) -> str:
        """Create a .tar.gz package of the current working directory.

        Applies exclusion patterns from FlameContext.package.excludes.

        Returns:
            Path to the created package file

        Raises:
            FlameError: If package creation fails
        """
        cwd = os.getcwd()
        dist_dir = os.path.join(cwd, "dist")

        # Create dist directory if it doesn't exist
        os.makedirs(dist_dir, exist_ok=True)

        generated_pyproject = self._generated_pyproject_toml(cwd)

        package_filename = f"{self._name}.tar.gz"
        package_path = os.path.join(dist_dir, package_filename)

        default_excludes = [
            ".venv",
            "venv",
            "__pycache__",
            ".pytest_cache",
            ".ruff_cache",
            ".mypy_cache",
            "*.egg-info",
            ".git",
            ".tox",
            "node_modules",
            "*.pyc",
            "*.pyo",
            ".DS_Store",
        ]

        user_excludes = self._context.package.excludes if self._context.package else []
        excludes = list(set(default_excludes + user_excludes))

        logger.debug(f"Creating package with excludes: {excludes}")

        try:
            with tarfile.open(package_path, "w:gz") as tar:
                # Add files while respecting exclusions
                for item in os.listdir(cwd):
                    # Skip the dist directory (where the package is created)
                    if item == "dist":
                        continue

                    # Check if item matches any exclusion pattern
                    if self._should_exclude(item, excludes):
                        logger.debug(f"Excluding: {item}")
                        continue

                    item_path = os.path.join(cwd, item)
                    tar.add(item_path, arcname=item, recursive=True, filter=lambda tarinfo: None if self._should_exclude(tarinfo.name, excludes) else tarinfo)

                if generated_pyproject is not None:
                    data = generated_pyproject.encode("utf-8")
                    tarinfo = tarfile.TarInfo("pyproject.toml")
                    tarinfo.size = len(data)
                    tarinfo.mode = 0o644
                    tar.addfile(tarinfo, io.BytesIO(data))

            logger.debug(f"Created package: {package_path}")
            return package_path

        except Exception as e:
            raise FlameError(FlameErrorCode.INTERNAL, f"Failed to create package: {str(e)}")

    def _should_exclude(self, name: str, patterns: List[str]) -> bool:
        import fnmatch

        for pattern in patterns:
            if fnmatch.fnmatch(name, pattern) or fnmatch.fnmatch(os.path.basename(name), pattern):
                return True
        return False

    def _generated_pyproject_toml(self, cwd: str) -> Optional[str]:
        """Return generated package metadata when the source tree needs it."""
        if os.path.exists(os.path.join(cwd, "pyproject.toml")):
            logger.debug("pyproject.toml already exists, skipping generated metadata")
            return None

        has_legacy_metadata = os.path.exists(os.path.join(cwd, "setup.py")) or os.path.exists(os.path.join(cwd, "setup.cfg"))
        if has_legacy_metadata:
            if self._dependencies:
                logger.warning(
                    "Python package metadata (setup.py/setup.cfg) already exists. "
                    "Skipping pyproject.toml generation to avoid conflicting with existing metadata. "
                    "Please specify dependencies in your setup.py or setup.cfg."
                )
            else:
                logger.debug("Python package metadata already exists, skipping generated metadata")
            return None

        deps = sorted(self._dependencies or [])
        if deps:
            deps_block = "dependencies = [\n    " + ",\n    ".join(f'"{dep}"' for dep in deps) + ",\n]"
        else:
            deps_block = "dependencies = []"

        content = f'''[build-system]
requires = ["setuptools>=61.0", "wheel"]
build-backend = "setuptools.build_meta"

[project]
name = "{self._name}"
version = "0.1.0"
requires-python = ">=3.9"
{deps_block}

[tool.setuptools]
py-modules = []
'''

        logger.info(f"Generated package pyproject.toml with dependencies: {deps}")
        return content

    def _upload_package(self) -> str:
        """Upload the package to the storage location.

        Uses the configured storage backend to upload the package.

        Returns:
            The full URL to the uploaded package

        Raises:
            FlameError: If upload fails
        """
        if not self._package_path:
            raise FlameError(FlameErrorCode.INVALID_STATE, "Package path is not set")

        if not self._storage_backend:
            raise FlameError(FlameErrorCode.INVALID_STATE, "Storage backend is not initialized")

        package_filename = os.path.basename(self._package_path)
        return self._storage_backend.upload(self._package_path, package_filename)

    def _cleanup_storage(self) -> None:
        """Delete the package from storage."""
        if not self._package_path or not self._storage_backend:
            return

        try:
            package_filename = os.path.basename(self._package_path)
            self._storage_backend.delete(package_filename)
        except Exception as e:
            logger.error(f"Error cleaning up storage: {e}", exc_info=True)