Merge pull request #83 from labthings/blob_input

Blob input

Author: rwb27

src/labthings_fastapi/actions/__init__.py

@@ -10,8 +10,8 @@
 from fastapi import FastAPI, HTTPException, Request
 from fastapi.responses import FileResponse
 from pydantic import BaseModel
-from labthings_fastapi.outputs.blob import blob_to_link
 
+from labthings_fastapi.utilities import model_to_dict
 from labthings_fastapi.utilities.introspection import EmptyInput
 from ..thing_description.model import LinkElement
 from ..file_manager import FileManager
@@ -21,6 +21,7 @@
     InvocationCancelledError,
     invocation_logger,
 )
+from ..outputs.blob import BlobIOContextDep
 
 if TYPE_CHECKING:
     # We only need these imports for type hints, so this avoids circular imports.
@@ -158,32 +159,32 @@ def response(self, request: Optional[Request] = None):
             timeCompleted=self._end_time,
             timeRequested=self._request_time,
             input=self.input,
-            output=blob_to_link(self.output, href + "/output"),
+            output=self.output,
             links=links,
             log=self.log,
         )
 
     def run(self):
         """Overrides default threading.Thread run() method"""
-        self.action.emit_changed_event(self.thing, self._status)
+        try:
+            self.action.emit_changed_event(self.thing, self._status)
 
-        # Capture just this thread's log messages
-        handler = DequeLogHandler(dest=self._log)
-        logger = invocation_logger(self.id)
-        logger.addHandler(handler)
+            # Capture just this thread's log messages
+            handler = DequeLogHandler(dest=self._log)
+            logger = invocation_logger(self.id)
+            logger.addHandler(handler)
 
-        action = self.action
-        thing = self.thing
-        kwargs = self.input.model_dump() or {}
-        assert action is not None
-        assert thing is not None
+            action = self.action
+            thing = self.thing
+            kwargs = model_to_dict(self.input)
+            assert action is not None
+            assert thing is not None
 
-        with self._status_lock:
-            self._status = InvocationStatus.RUNNING
-            self._start_time = datetime.datetime.now()
-            self.action.emit_changed_event(self.thing, self._status)
+            with self._status_lock:
+                self._status = InvocationStatus.RUNNING
+                self._start_time = datetime.datetime.now()
+                self.action.emit_changed_event(self.thing, self._status)
 
-        try:
             # The next line actually runs the action.
             ret = action.__get__(thing)(**kwargs, **self.dependencies)
 
@@ -283,6 +284,11 @@ def invoke_action(
         thread.start()
         return thread
 
+    def get_invocation(self, id: uuid.UUID) -> Invocation:
+        """Retrieve an invocation by ID"""
+        with self._invocations_lock:
+            return self._invocations[id]
+
     def list_invocations(
         self,
         action: Optional[ActionDescriptor] = None,
@@ -314,15 +320,17 @@ def attach_to_app(self, app: FastAPI):
         """Add /action_invocations and /action_invocation/{id} endpoints to FastAPI"""
 
         @app.get(ACTION_INVOCATIONS_PATH, response_model=list[InvocationModel])
-        def list_all_invocations(request: Request):
+        def list_all_invocations(request: Request, _blob_manager: BlobIOContextDep):
             return self.list_invocations(as_responses=True, request=request)
 
         @app.get(
             ACTION_INVOCATIONS_PATH + "/{id}",
             response_model=InvocationModel,
             responses={404: {"description": "Invocation ID not found"}},
         )
-        def action_invocation(id: uuid.UUID, request: Request):
+        def action_invocation(
+            id: uuid.UUID, request: Request, _blob_manager: BlobIOContextDep
+        ):
             try:
                 with self._invocations_lock:
                     return self._invocations[id].response(request=request)
@@ -346,7 +354,7 @@ def action_invocation(id: uuid.UUID, request: Request):
                 503: {"description": "No result is available for this invocation"},
             },
         )
-        def action_invocation_output(id: uuid.UUID):
+        def action_invocation_output(id: uuid.UUID, _blob_manager: BlobIOContextDep):
             """Get the output of an action invocation
 
             This returns just the "output" component of the action invocation. If the

src/labthings_fastapi/client/__init__.py

@@ -72,6 +72,10 @@ def set_property(self, path: str, value: Any):
         r.raise_for_status()
 
     def invoke_action(self, path: str, **kwargs):
+        "Invoke an action on the Thing"
+        for k in kwargs.keys():
+            if isinstance(kwargs[k], ClientBlobOutput):
+                kwargs[k] = {"href": kwargs[k].href, "media_type": kwargs[k].media_type}
         r = self.client.post(urljoin(self.path, path), json=kwargs)
         r.raise_for_status()
         task = poll_task(self.client, r.json())
@@ -118,14 +122,15 @@ def from_url(
     @classmethod
     def subclass_from_td(cls, thing_description: dict) -> type[Self]:
         """Create a ThingClient subclass from a Thing Description"""
+        my_thing_description = thing_description
 
         class Client(cls):  # type: ignore[valid-type, misc]
             # mypy wants the superclass to be statically type-able, but
             # this isn't possible (for now) if we are to be able to
             # use this class method on `ThingClient` subclasses, i.e.
             # to provide customisation but also add methods from a
             # Thing Description.
-            pass
+            thing_description = my_thing_description
 
         for name, p in thing_description["properties"].items():
             add_property(Client, name, p)

src/labthings_fastapi/client/in_server.py

@@ -20,7 +20,7 @@
 from labthings_fastapi.utilities import attributes
 from . import PropertyClientDescriptor
 from ..thing import Thing
-from ..server import find_thing_server
+from ..dependencies.thing_server import find_thing_server
 from fastapi import Request
 
 

src/labthings_fastapi/dependencies/action_manager.py

@@ -0,0 +1,57 @@
+"""
+Context Var access to the Action Manager
+
+This module provides a context var to access the Action Manager instance.
+While LabThings tries pretty hard to conform to FastAPI's excellent convention
+that everything should be passed as a parameter, there are some cases where
+that's hard. In particular, generating URLs when responses are serialised is
+difficult, because `pydantic` doesn't have a way to pass in extra context.
+
+If an endpoint uses the `ActionManagerDep` dependency, then the Action Manager
+is available using `ActionManagerContext.get()`.
+"""
+
+from __future__ import annotations
+
+from contextvars import ContextVar
+
+from typing import Annotated
+from typing_extensions import TypeAlias
+from fastapi import Depends, Request
+from ..dependencies.thing_server import find_thing_server
+from ..actions import ActionManager
+
+
+def action_manager_from_thing_server(request: Request) -> ActionManager:
+    """Retrieve the Action Manager from the Thing Server
+
+    This is for use as a FastAPI dependency, so the thing server is
+    retrieved from the request object.
+    """
+    action_manager = find_thing_server(request.app).action_manager
+    if action_manager is None:
+        raise RuntimeError("Could not get the blocking portal from the server.")
+    return action_manager
+
+
+ActionManagerDep = Annotated[ActionManager, Depends(action_manager_from_thing_server)]
+"""
+A ready-made dependency type for the `ActionManager` object.
+"""
+
+
+ActionManagerContext = ContextVar[ActionManager]("ActionManagerContext")
+
+
+async def make_action_manager_context_available(action_manager: ActionManagerDep):
+    """Make the Action Manager available in the context
+
+    The action manager may be accessed using `ActionManagerContext.get()`.
+    """
+    ActionManagerContext.set(action_manager)
+    yield action_manager
+
+
+ActionManagerContextDep: TypeAlias = Annotated[
+    ActionManager, Depends(make_action_manager_context_available)
+]

src/labthings_fastapi/dependencies/blocking_portal.py

@@ -8,14 +8,14 @@
 from typing import Annotated
 from fastapi import Depends, Request
 from anyio.from_thread import BlockingPortal as RealBlockingPortal
-from ..server import find_thing_server
+from .thing_server import find_thing_server
 
 
 def blocking_portal_from_thing_server(request: Request) -> RealBlockingPortal:
-    """Return a UUID for an action invocation
+    """Return the blocking portal from our ThingServer
 
-    This is for use as a FastAPI dependency, to allow other dependencies to
-    access the invocation ID. Useful for e.g. file management.
+    This is for use as a FastAPI dependency, to allow threaded code to call
+    async code.
     """
     portal = find_thing_server(request.app).blocking_portal
     if portal is None:

src/labthings_fastapi/dependencies/metadata.py

@@ -4,7 +4,7 @@
 
 from fastapi import Depends, Request
 
-from ..server import find_thing_server
+from .thing_server import find_thing_server
 
 
 def thing_states_getter(request: Request) -> Callable[[], Mapping[str, Any]]:

src/labthings_fastapi/dependencies/raw_thing.py

@@ -4,7 +4,7 @@
 from fastapi import Depends, Request
 
 from ..thing import Thing
-from ..server import find_thing_server
+from .thing_server import find_thing_server
 
 
 ThingInstance = TypeVar("ThingInstance", bound=Thing)

src/labthings_fastapi/dependencies/thing_server.py

@@ -0,0 +1,34 @@
+"""
+Retrieve the ThingServer object
+
+This module provides a function that will retrieve the ThingServer
+based on the `Request` object. It may be used as a dependency with:
+`Annotated[ThingServer, Depends(thing_server_from_request)]`.
+"""
+
+from __future__ import annotations
+from weakref import WeakSet
+from typing import TYPE_CHECKING
+from fastapi import FastAPI, Request
+
+if TYPE_CHECKING:
+    from labthings_fastapi.server import ThingServer
+
+_thing_servers: WeakSet[ThingServer] = WeakSet()
+
+
+def find_thing_server(app: FastAPI) -> ThingServer:
+    """Find the ThingServer associated with an app"""
+    for server in _thing_servers:
+        if server.app == app:
+            return server
+    raise RuntimeError("No ThingServer found for this app")
+
+
+def thing_server_from_request(request: Request) -> ThingServer:
+    """Retrieve the Action Manager from the Thing Server
+
+    This is for use as a FastAPI dependency, so the thing server is
+    retrieved from the request object.
+    """
+    return find_thing_server(request.app)

src/labthings_fastapi/descriptors/action.py

@@ -10,6 +10,7 @@
 from pydantic import create_model
 from ..actions import InvocationModel
 from ..dependencies.invocation import CancelHook, InvocationID
+from ..dependencies.action_manager import ActionManagerContextDep
 from ..utilities.introspection import (
     EmptyInput,
     StrictEmptyInput,
@@ -19,7 +20,7 @@
     input_model_from_signature,
     return_type,
 )
-from ..outputs.blob import blob_to_model, get_model_media_type
+from ..outputs.blob import BlobIOContextDep
 from ..thing_description import type_to_dataschema
 from ..thing_description.model import ActionAffordance, ActionOp, Form, Union
 
@@ -69,7 +70,7 @@ def __init__(
             remove_first_positional_arg=True,
             ignore=[p.name for p in self.dependency_params],
         )
-        self.output_model = blob_to_model(return_type(func))
+        self.output_model = return_type(func)
         self.invocation_model = create_model(
             f"{self.name}_invocation",
             __base__=InvocationModel,
@@ -163,6 +164,8 @@ def add_to_fastapi(self, app: FastAPI, thing: Thing):
         # The solution below is to manually add the annotation, before passing
         # the function to the decorator.
         def start_action(
+            action_manager: ActionManagerContextDep,
+            _blob_manager: BlobIOContextDep,
             request: Request,
             body,
             id: InvocationID,
@@ -171,15 +174,15 @@ def start_action(
             **dependencies,
         ):
             try:
-                action = thing.action_manager.invoke_action(
+                action = action_manager.invoke_action(
                     action=self,
                     thing=thing,
                     input=body,
                     dependencies=dependencies,
                     id=id,
                     cancel_hook=cancel_hook,
                 )
-                background_tasks.add_task(thing.action_manager.expire_invocations)
+                background_tasks.add_task(action_manager.expire_invocations)
                 return action.response(request=request)
             finally:
                 try:
@@ -219,8 +222,8 @@ def start_action(
         except AttributeError:
             print(f"Failed to generate response model for action {self.name}")
         # Add an additional media type if we may return a file
-        if get_model_media_type(self.output_model):
-            responses[200]["content"][get_model_media_type(self.output_model)] = {}
+        if hasattr(self.output_model, "media_type"):
+            responses[200]["content"][self.output_model.media_type] = {}
         # Now we can add the endpoint to the app.
         app.post(
             thing.path + self.name,
@@ -243,8 +246,8 @@ def start_action(
             ),
             summary=f"All invocations of {self.name}.",
         )
-        def list_invocations():
-            return thing.action_manager.list_invocations(self, thing, as_responses=True)
+        def list_invocations(action_manager: ActionManagerContextDep):
+            return action_manager.list_invocations(self, thing, as_responses=True)
 
     def action_affordance(
         self, thing: Thing, path: Optional[str] = None

src/labthings_fastapi/file_manager.py

@@ -1,5 +1,7 @@
 """Manage files created by Actions
 
+**This module is deprecated in favour of `labthings_fastapi.outputs.blob`**
+
 Simple actions return everything you need to know about them in their return value,
 which can be serialised to JSON. More complicated actions might need to return
 more complicated responses, for example files.
@@ -12,6 +14,7 @@
 from __future__ import annotations
 from tempfile import TemporaryDirectory
 from typing import Annotated, Sequence, Optional
+from warnings import warn
 
 from fastapi import Depends, Request
 
@@ -27,6 +30,10 @@ class FileManager:
     __globals__ = globals()  # "bake in" globals so dependency injection works
 
     def __init__(self, invocation_id: InvocationID, request: Request):
+        warn(
+            "FileManager is deprecated in favour of labthings_fastapi.outputs.blob",
+            DeprecationWarning,
+        )
         self.invocation_id = invocation_id
         self._links: set[tuple[str, str]] = set()
         self._dir = TemporaryDirectory(prefix=f"labthings-{self.invocation_id}-")