Method execute

src/zenml/zen_server/request_management.py:543–669

Execute a request with in-memory de-duplication. Call this method to execute a request with in-memory de-duplication. If the request (identified by the request_id) has previously been executed and the result is still cached, the result will be returned immediately. I

(
        self,
        func: Callable[..., Any],
        deduplicate: Optional[bool],
        *args: Any,
        **kwargs: Any,
    )

            request_record.set_result(result)

    async def execute(
        self,
        func: Callable[..., Any],
        deduplicate: Optional[bool],
        *args: Any,
        **kwargs: Any,
    ) -> Any:
        """Execute a request with in-memory de-duplication.

        Call this method to execute a request with in-memory de-duplication.
        If the request (identified by the request_id) has previously been
        executed and the result is still cached, the result will be returned
        immediately. If the request is in-flight, the method will wait for it
        to complete. If the request is not in-flight, the method will start
        execution in the background and cache the result.

        Args:
            func: The function to execute.
            deduplicate: Whether to enable or disable request deduplication for
                this request. If not specified, by default, the deduplication
                is enabled for POST requests and disabled for other requests.
            *args: The arguments to pass to the function.
            **kwargs: The keyword arguments to pass to the function.

        Returns:
            The result of the request.
        """
        from zenml.zen_server.utils import get_system_metrics

        request_context = self.current_request
        assert request_context is not None

        logger.debug(
            "endpoint.started",
            extra={
                "endpoint": func.__name__,
                **get_system_metrics(),
            },
        )

        transaction_id = request_context.transaction_id

        if deduplicate is None:
            # If not specified, by default, the deduplication is enabled for
            # POST requests and disabled for other requests.
            deduplicate = request_context.request.method == "POST"

        deduplicate = (
            deduplicate and self.deduplicate and request_context.is_cacheable
        )

        async with self.lock:
            if deduplicate and transaction_id in self.transactions:
                # The transaction is still being processed on the same
                # server instance. We just wait for it to complete.
                fut = self.transactions[transaction_id].future
                logger.debug(
                    "endpoint.resumed",