API Reference

公开 API

blendjob

Reusable local HTTP Job Server for Blender add-ons.

JobResult dataclass

Successful Job value and files returned to a Blender Operator.

源代码位于： src\blendjob\client.py

@dataclass(frozen=True)
class JobResult:
    """Successful Job value and files returned to a Blender Operator."""

    job_id: str
    directory: Path
    value: object

    @classmethod
    def _from_status(cls, status):
        return cls(
            job_id=str(status.get("job_id", "")),
            directory=Path(status["directory"]),
            value=status.get("result"),
        )

    def file(self, name):
        if not isinstance(self.value, dict):
            raise RuntimeError("Job result does not contain named files")
        try:
            relative = self.value[name]
        except KeyError:
            raise RuntimeError(f"Job result does not contain file: {name}") from None
        root = self.directory.resolve()
        path = (root / str(relative)).resolve()
        if path != root and root not in path.parents:
            raise RuntimeError(f"Job result file escapes its directory: {relative}")
        if not path.is_file():
            raise RuntimeError(f"Job result file does not exist: {path}")
        return path

JobRuntime

Own one add-on's Storage Root, Environment, Server and Blender UI.

源代码位于： src\blendjob\runtime.py

class JobRuntime:
    """Own one add-on's Storage Root, Environment, Server and Blender UI."""

    def __init__(
        self,
        server_entrypoint,
        *,
        entrypoint_root=None,
        storage_root,
        environment,
        namespace,
        post_install=None,
    ):
        self.server_entrypoint = normalized_entrypoint(
            server_entrypoint,
            root=entrypoint_root,
        )
        self.storage_root_factory = storage_root
        self.environment = normalized_environment(environment)
        self.environment_hash = environment_digest(self.environment)
        if post_install is not None and not callable(post_install):
            raise TypeError("post_install must be callable")
        self.post_install = post_install
        if not isinstance(namespace, str) or not namespace.strip():
            raise ValueError("namespace must be a non-empty string")
        self.namespace = namespace.strip()
        self.active_job = None
        self.progress = 0.0
        self.message = "Ready"
        self._operator_base = None
        self._operator_classes = None
        self._timer_registered = False
        self._registered = False
        self._status_bar_draw = self._create_status_bar_draw()
        self._request_progress = None
        self._request_cancel_check = None
        self._environment_controller = EnvironmentController(self)
        self.server = ServerController(
            self._server_command,
            cwd_factory=self.storage_root,
            log_path_factory=self.server_log_path,
            creationflags=windows_creation_flags(),
        )
        self.JobOperatorBase = self.operator_base()

    def storage_root(self):
        configured = (
            self.storage_root_factory()
            if callable(self.storage_root_factory)
            else self.storage_root_factory
        )
        root = Path(configured).expanduser().resolve()
        root.mkdir(parents=True, exist_ok=True)
        return root

    def environment_directory(self):
        return self.storage_root() / ".venv"

    def environment_python(self):
        directory = self.environment_directory()
        if os.name == "nt":
            return directory / "Scripts" / "python.exe"
        return directory / "bin" / "python"

    def environment_manifest_path(self):
        return self.storage_root() / "manifest.json"

    def environment_config_path(self):
        return self.storage_root() / "environment.json"

    def install_status_path(self):
        return self.storage_root() / "install-status.json"

    def install_log_path(self):
        return self.storage_root() / "install.log"

    def server_log_path(self):
        return self.storage_root() / "server.log"

    def environment_manifest(self):
        try:
            value = json.loads(
                self.environment_manifest_path().read_text(encoding="utf-8")
            )
        except (FileNotFoundError, json.JSONDecodeError, OSError):
            return {}
        return value if isinstance(value, dict) else {}

    def environment_ready(self):
        return (
            self.environment_python().is_file()
            and self.environment_manifest().get("environment_hash")
            == self.environment_hash
        )

    def write_environment_config(self):
        path = self.environment_config_path()
        path.write_text(
            json.dumps(self.environment, indent=2),
            encoding="utf-8",
        )
        return path

    def install_command(self, source="official"):
        installer = Path(__file__).with_name("installer") / "environment.py"
        config = self.write_environment_config()
        return [
            sys.executable,
            str(installer),
            "--storage-root",
            str(self.storage_root()),
            "--config",
            str(config),
            "--status",
            str(self.install_status_path()),
            "--source",
            source,
        ]

    def request(self, job_type, parameters, response=None):
        """Directly request one Server Job and return its JobResult."""
        if self._request_cancel_check is not None and self._request_cancel_check():
            raise RuntimeError("Server request was cancelled")
        return self.server.request(
            job_type,
            parameters,
            response=response,
            progress=self._request_progress,
            cancel_check=self._request_cancel_check,
        )

    def operator_base(self):
        if self._operator_base is not None:
            return self._operator_base
        runtime = self

        class BoundJobOperator(JobOperatorBase):
            job_runtime = runtime

        BoundJobOperator.__name__ = "JobOperatorBase"
        self._operator_base = BoundJobOperator
        return BoundJobOperator

    def operator_classes(self):
        if self._operator_classes is not None:
            return self._operator_classes
        import bpy

        runtime = self
        operator_base = self.JobOperatorBase

        class InstallEnvironment(operator_base, bpy.types.Operator):
            bl_idname = f"{runtime.namespace}.install_environment"
            bl_label = "Install Environment"
            job_type = "install-environment"
            starting_message = "Starting installation ..."

            def request(self, _context):
                if not bpy.app.online_access:
                    raise RuntimeError(
                        "Online Access is disabled in Blender Preferences"
                    )
                return {"source": "official"}

            def controller(self, _runtime):
                runtime.server.stop()
                return runtime._environment_controller

        class InstallEnvironmentMirror(operator_base, bpy.types.Operator):
            bl_idname = f"{runtime.namespace}.install_environment_mirror"
            bl_label = "Install Environment (Mirror)"
            job_type = "install-environment"
            starting_message = "Starting installation ..."

            def request(self, _context):
                if not bpy.app.online_access:
                    raise RuntimeError(
                        "Online Access is disabled in Blender Preferences"
                    )
                return {"source": "mirror"}

            def controller(self, _runtime):
                runtime.server.stop()
                return runtime._environment_controller

        class CancelJob(bpy.types.Operator):
            bl_idname = f"{runtime.namespace}.cancel_job"
            bl_label = "Cancel Job"

            @classmethod
            def poll(cls, _context):
                return runtime.active_job is not None

            def execute(self, _context):
                runtime.cancel_active()
                return {"FINISHED"}

        class StartServer(bpy.types.Operator):
            bl_idname = f"{runtime.namespace}.start_server"
            bl_label = "Start Server"

            def execute(self, _context):
                try:
                    runtime.server.enable()
                except (OSError, RuntimeError) as error:
                    self.report({"ERROR"}, str(error))
                    return {"CANCELLED"}
                return {"FINISHED"}

        class StopServer(bpy.types.Operator):
            bl_idname = f"{runtime.namespace}.stop_server"
            bl_label = "Stop Server"

            def execute(self, _context):
                runtime.server.stop()
                runtime.redraw_ui()
                return {"FINISHED"}

        class RestartServer(bpy.types.Operator):
            bl_idname = f"{runtime.namespace}.restart_server"
            bl_label = "Restart Server"

            def execute(self, _context):
                try:
                    runtime.server.restart()
                except (OSError, RuntimeError) as error:
                    self.report({"ERROR"}, str(error))
                    return {"CANCELLED"}
                return {"FINISHED"}

        class OpenServerLog(bpy.types.Operator):
            bl_idname = f"{runtime.namespace}.open_server_log"
            bl_label = "Open Server Log"

            def execute(self, _context):
                path = runtime.server_log_path()
                target = path if path.is_file() else path.parent
                result = bpy.ops.wm.path_open(filepath=str(target))
                if "FINISHED" not in result:
                    self.report({"ERROR"}, "Unable to open the Server log")
                    return {"CANCELLED"}
                return {"FINISHED"}

        self._operator_classes = (
            InstallEnvironment,
            InstallEnvironmentMirror,
            CancelJob,
            StartServer,
            StopServer,
            RestartServer,
            OpenServerLog,
        )
        return self._operator_classes

    def register(self):
        if self._registered:
            return
        import bpy

        for class_type in self.operator_classes():
            bpy.utils.register_class(class_type)
        bpy.types.STATUSBAR_HT_header.append(self._status_bar_draw)
        self._registered = True
        self.enable()

    def unregister(self):
        if not self._registered:
            return
        import bpy

        self.disable()
        bpy.types.STATUSBAR_HT_header.remove(self._status_bar_draw)
        for class_type in reversed(self.operator_classes()):
            bpy.utils.unregister_class(class_type)
        self._registered = False

    def begin_job(self, job):
        if self.active_job is not None:
            raise RuntimeError("Another task is running")
        self.active_job = job

    def finish_job(self, job):
        if job.job_id:
            job.controller.mark_job_complete(job.job_id)
        try:
            job.operator.cleanup()
        finally:
            if self.active_job is job:
                self.active_job = None

    def cancel_active(self):
        job = self.active_job
        if job is None:
            return
        job.cancelled = True
        if not job.started or not job.job_id:
            return
        try:
            job.controller.cancel(job.job_id)
        except RuntimeError:
            pass

    def close_active(self):
        job = self.active_job
        if job is not None:
            self.finish_job(job)

    def update_ui(self, context, progress, message):
        self.progress = min(max(float(progress), 0.0), 1.0)
        self.message = str(message)
        window_manager = getattr(context, "window_manager", None)
        if window_manager is not None and hasattr(
            window_manager, "progress_update"
        ):
            window_manager.progress_update(int(self.progress * 100))
        self.redraw_ui(context)

    def redraw_ui(self, context=None, force=False):
        try:
            import bpy
        except ModuleNotFoundError:
            return
        context = context or bpy.context
        window_manager = getattr(context, "window_manager", None)
        for window in getattr(window_manager, "windows", ()):
            screen = getattr(window, "screen", None)
            for area in getattr(screen, "areas", ()):
                if area.type in {"STATUSBAR", "VIEW_3D", "PREFERENCES"}:
                    area.tag_redraw()
                    for region in getattr(area, "regions", ()):
                        region.tag_redraw()
        if not force:
            return
        workspace = getattr(context, "workspace", None)
        if workspace is not None:
            try:
                workspace.status_text_set_internal(None)
            except (AttributeError, RuntimeError, TypeError):
                pass

    def server_status(self):
        return dict(self.server.snapshot)

    def server_busy(self):
        return self.active_job is not None or self.server.snapshot.get(
            "state"
        ) == "BUSY"

    def resource(self, name):
        return self.server.resource(name)

    def clear_resource(self, name):
        return self.server.clear_resource(name)

    def enable(self):
        try:
            import bpy
        except ModuleNotFoundError:
            return
        timers = getattr(bpy.app, "timers", None)
        if timers is None or timers.is_registered(self._poll_server):
            return
        timers.register(self._poll_server, first_interval=1.0, persistent=True)
        self._timer_registered = True

    def disable(self):
        self.cancel_active()
        self.close_active()
        try:
            import bpy
        except ModuleNotFoundError:
            bpy = None
        timers = getattr(getattr(bpy, "app", None), "timers", None)
        if timers is not None and timers.is_registered(self._poll_server):
            timers.unregister(self._poll_server)
        self._timer_registered = False
        self.server.detach()

    def _poll_server(self):
        interval = self.server.poll(available=self.environment_ready())
        self.redraw_ui()
        return interval

    def _draw_status_bar(self, owner, _context):
        if self.active_job is None:
            return
        row = owner.layout.row(align=True)
        progress_row = row.row(align=True)
        progress_row.ui_units_x = 18.0
        progress_row.progress(
            factor=self.progress,
            type="BAR",
            text=self.message,
        )
        row.operator(
            f"{self.namespace}.cancel_job",
            text="",
            icon="X",
        )

    def _create_status_bar_draw(self):
        runtime = self

        def draw_status_bar(owner, context):
            runtime._draw_status_bar(owner, context)

        return draw_status_bar

    def _server_command(self, port, instance_id):
        launcher = Path(__file__).with_name("launcher") / "runner.py"
        return [
            str(self.environment_python()),
            "-u",
            str(launcher),
            "--entrypoint",
            self.server_entrypoint,
            "--storage-root",
            str(self.storage_root()),
            "--host",
            "127.0.0.1",
            "--port",
            str(port),
            "--instance-id",
            instance_id,
            "--parent-pid",
            str(os.getpid()),
        ]

request(job_type, parameters, response=None)

Directly request one Server Job and return its JobResult.

源代码位于： src\blendjob\runtime.py

def request(self, job_type, parameters, response=None):
    """Directly request one Server Job and return its JobResult."""
    if self._request_cancel_check is not None and self._request_cancel_check():
        raise RuntimeError("Server request was cancelled")
    return self.server.request(
        job_type,
        parameters,
        response=response,
        progress=self._request_progress,
        cancel_check=self._request_cancel_check,
    )

JobContext

Mutable state and cancellation API passed to a registered job handler.

源代码位于： src\blendjob\server.py

class JobContext:
    """Mutable state and cancellation API passed to a registered job handler."""

    def __init__(
        self,
        job_id,
        job_type,
        parameters,
        resources,
        storage_root,
        directory,
    ):
        self.job_id = job_id
        self.job_type = job_type
        self._parameters = parameters
        self.storage_root = Path(storage_root)
        self.directory = Path(directory)
        self._resources = resources
        self._lock = threading.Lock()
        self._cancel_requested = False
        self._status = {
            "job_id": job_id,
            "job_type": job_type,
            "state": "queued",
            "progress": 0.0,
            "message": "Task queued",
            "directory": str(self.directory),
        }

    def _update(
        self,
        progress,
        message,
        *,
        state,
        error=None,
        result=_NO_RESULT,
    ):
        snapshot = {
            "job_id": self.job_id,
            "job_type": self.job_type,
            "state": state,
            "progress": min(max(float(progress), 0.0), 1.0),
            "message": str(message),
            "directory": str(self.directory),
        }
        if error:
            snapshot["error"] = str(error)
        if result is not _NO_RESULT:
            snapshot["result"] = result
        with self._lock:
            self._status = snapshot

    def progress(self, progress, message):
        """Publish running progress from a Job handler."""
        self._update(progress, message, state="running")

    def _succeed(self, result=None, message="Task complete"):
        self._update(1.0, message, state="succeeded", result=result)

    def _snapshot(self):
        with self._lock:
            return dict(self._status)

    def _request_cancel(self):
        with self._lock:
            if self._status["state"] in TERMINAL_STATES:
                return False
            self._cancel_requested = True
            self._status = {
                **self._status,
                "state": "cancelling",
                "message": "Cancelling task",
            }
            return True

    def _is_cancelled(self):
        with self._lock:
            return self._cancel_requested

    def check_cancelled(self):
        if self._is_cancelled():
            raise JobCancelled("Task cancelled")

    def resource(self, name):
        """Return a shared resource registered on the owning JobServer."""
        try:
            return self._resources[name]
        except KeyError:
            raise KeyError(f"Unknown Server Resource: {name}") from None

progress(progress, message)

Publish running progress from a Job handler.

源代码位于： src\blendjob\server.py

def progress(self, progress, message):
    """Publish running progress from a Job handler."""
    self._update(progress, message, state="running")

resource(name)

Return a shared resource registered on the owning JobServer.

源代码位于： src\blendjob\server.py

def resource(self, name):
    """Return a shared resource registered on the owning JobServer."""
    try:
        return self._resources[name]
    except KeyError:
        raise KeyError(f"Unknown Server Resource: {name}") from None

JobServer

Small FastAPI wrapper with decorator-based job registration.

源代码位于： src\blendjob\server.py

class JobServer:
    """Small FastAPI wrapper with decorator-based job registration."""

    def __init__(
        self,
        name,
        *,
        storage_root=None,
        max_job_history=100,
    ):
        self.name = name
        self.max_job_history = max(int(max_job_history), 0)
        self.handlers = {}
        self.resources = {}
        self.resource_factories = {}
        self.jobs = {}
        self.futures = {}
        self.active_job_id = None
        self.lock = threading.Lock()
        self.executor = ThreadPoolExecutor(
            max_workers=1,
            thread_name_prefix="BlendJobWorker",
        )
        self.closed = False
        self.shutdown_event = threading.Event()
        self.started_at = time.time()
        self.storage_root = Path(storage_root) if storage_root is not None else None
        self.jobs_directory = (
            self.storage_root / "jobs" if self.storage_root is not None else None
        )

    def bind(self, storage_root):
        """Bind the Server to the Runtime-owned Storage Root."""
        storage_root = Path(storage_root).resolve()
        if self.storage_root is not None:
            current = self.storage_root.resolve()
            if current != storage_root:
                raise RuntimeError(
                    f"Job Server is already bound to another Storage Root: {current}"
                )
        self.storage_root = storage_root
        self.jobs_directory = storage_root / "jobs"
        for name, factory in tuple(self.resource_factories.items()):
            if name not in self.resources:
                self.resources[name] = factory(self)
        return self

    def add_resource(self, name, resource):
        """Register shared state owned for the full Server lifetime."""
        if not name or "/" in name:
            raise ValueError(f"Invalid resource name: {name}")
        with self.lock:
            if name in self.resources:
                raise ValueError(f"Resource is already registered: {name}")
            self.resources[name] = resource
        return resource

    def resource(self, name):
        """Register a Resource factory initialized after Storage Root binding."""
        def register(factory):
            if not name or "/" in name:
                raise ValueError(f"Invalid resource name: {name}")
            if name in self.resources or name in self.resource_factories:
                raise ValueError(f"Resource is already registered: {name}")
            self.resource_factories[name] = factory
            if self.storage_root is not None:
                self.resources[name] = factory(self)
            return factory

        return register

    def resource_snapshot(self, name):
        try:
            resource = self.resources[name]
        except KeyError:
            raise KeyError(name) from None
        snapshot = getattr(resource, "snapshot", None)
        return snapshot() if snapshot is not None else {}

    def resource_snapshots(self):
        return {
            name: self.resource_snapshot(name)
            for name in tuple(self.resources)
        }

    def clear_resource(self, name):
        """Clear an idle Server Resource without racing a new Job submission."""
        with self.lock:
            if self.active_job_id is not None or self._queued_job_count():
                return False
            try:
                resource = self.resources[name]
            except KeyError:
                raise KeyError(name) from None
            clear = getattr(resource, "clear", None)
            if clear is None:
                raise TypeError(f"Resource cannot be cleared: {name}")
            clear()
            return True

    def close(self):
        with self.lock:
            if self.closed:
                return
            self.closed = True
            self.shutdown_event.set()
        self.executor.shutdown(wait=True, cancel_futures=True)
        for resource in reversed(tuple(self.resources.values())):
            close = getattr(resource, "close", None)
            if close is not None:
                close()

    def job(self, job_type):
        """Register ``handler(context, parameters)`` for a public job type."""
        def register(handler):
            if job_type in self.handlers:
                raise ValueError(f"Job type is already registered: {job_type}")
            self.handlers[job_type] = handler
            return handler

        return register

    def snapshot(self, instance_id):
        with self.lock:
            active = self.jobs.get(self.active_job_id)
            queued_jobs = self._queued_job_count()
            result = {
                "ready": True,
                "server": self.name,
                "instance_id": instance_id,
                "busy": active is not None or queued_jobs > 0,
                "active_job": active._snapshot() if active else None,
                "queued_jobs": queued_jobs,
                "started_at": self.started_at,
                "resources": self.resource_snapshots(),
            }
        return result

    def submit(self, job_type, parameters, *, job_id=None):
        if self.storage_root is None or self.jobs_directory is None:
            raise RuntimeError("Bind the Job Server to a Storage Root before submitting")
        job_id = job_id or uuid.uuid4().hex
        if job_type not in self.handlers:
            raise KeyError(job_type)
        if not job_id or Path(job_id).name != job_id or job_id in {".", ".."}:
            raise ValueError(f"Invalid job id: {job_id}")
        with self.lock:
            if self.closed:
                raise RuntimeError("Job Server is closed")
            if job_id in self.jobs:
                raise ValueError(f"Job id already exists: {job_id}")
            self.jobs_directory.mkdir(parents=True, exist_ok=True)
            directory = self.jobs_directory / job_id
            try:
                directory.mkdir()
            except FileExistsError:
                raise ValueError(f"Job directory already exists: {job_id}") from None
            context = JobContext(
                job_id,
                job_type,
                parameters,
                self.resources,
                self.storage_root,
                directory,
            )
            self.jobs[job_id] = context
            self._prune_jobs()
            self.futures[job_id] = self.executor.submit(self._run, context)
        return context

    def cancel(self, job_id):
        with self.lock:
            context = self.jobs.get(job_id)
            future = self.futures.get(job_id)
            if context is None:
                return None
            if future is not None and future.cancel():
                context._request_cancel()
                context._update(1.0, "Task cancelled", state="cancelled")
                self.futures.pop(job_id, None)
                return context
        return context if context._request_cancel() else None

    def _queued_job_count(self):
        return sum(
            context._snapshot()["state"] == "queued"
            for context in self.jobs.values()
        )

    def _prune_jobs(self):
        completed = [
            identifier
            for identifier, context in self.jobs.items()
            if identifier != self.active_job_id
            and context._snapshot()["state"] in TERMINAL_STATES
        ]
        excess = len(completed) - self.max_job_history
        for identifier in completed[:max(excess, 0)]:
            self.jobs.pop(identifier, None)

    def _run(self, context):
        with self.lock:
            if context._snapshot()["state"] in TERMINAL_STATES:
                return
            self.active_job_id = context.job_id
        context._update(0.0, "Server accepted the task", state="running")
        try:
            context.check_cancelled()
            result = self.handlers[context.job_type](
                context,
                context._parameters,
            )
            context.check_cancelled()
            snapshot = context._snapshot()
            if snapshot["state"] not in {"failed", "cancelled"}:
                context._succeed(
                    result,
                    message=snapshot.get("message") or "Task complete",
                )
        except JobCancelled:
            context._update(1.0, "Task cancelled", state="cancelled")
        except Exception as error:
            traceback.print_exc()
            context._update(1.0, "Task failed", error=error, state="failed")
        finally:
            with self.lock:
                if self.active_job_id == context.job_id:
                    self.active_job_id = None
                self.futures.pop(context.job_id, None)

    def create_app(self, instance_id, storage_root=None):
        from fastapi import FastAPI, HTTPException

        if storage_root is not None:
            self.bind(storage_root)
        elif self.storage_root is None:
            raise RuntimeError("Storage Root is required to create the Job Server app")

        @asynccontextmanager
        async def lifespan(_app):
            yield
            self.close()

        app = FastAPI(
            title=self.name,
            docs_url=None,
            redoc_url=None,
            openapi_url=None,
            lifespan=lifespan,
        )
        app.state.job_server = self

        @app.get("/health")
        def health():
            return self.snapshot(instance_id)

        @app.get("/resources")
        def resources():
            return self.resource_snapshots()

        @app.get("/resources/{resource_name}")
        def resource_status(
            resource_name: str,
        ):
            try:
                return self.resource_snapshot(resource_name)
            except KeyError:
                raise HTTPException(status_code=404, detail="Resource was not found")

        @app.post("/resources/{resource_name}/clear")
        def clear_resource(
            resource_name: str,
        ):
            try:
                cleared = self.clear_resource(resource_name)
            except KeyError:
                raise HTTPException(status_code=404, detail="Resource was not found")
            except TypeError as error:
                raise HTTPException(status_code=405, detail=str(error))
            if not cleared:
                raise HTTPException(status_code=409, detail="Server is busy")
            return {"resource": resource_name, "state": "cleared"}

        @app.post("/jobs", status_code=202)
        def submit_job(payload: dict):
            job_type = str(payload.get("job_type", ""))
            parameters = payload.get("parameters")
            if not job_type or not isinstance(parameters, dict):
                raise HTTPException(status_code=422, detail="Invalid job request")
            try:
                context = self.submit(job_type, parameters)
            except KeyError:
                raise HTTPException(status_code=422, detail="Unknown job type")
            except ValueError as error:
                raise HTTPException(status_code=409, detail=str(error))
            return context._snapshot()

        @app.get("/jobs/{job_id}")
        def job_status(job_id: str):
            context = self.jobs.get(job_id)
            if context is None:
                raise HTTPException(status_code=404, detail="Job was not found")
            return context._snapshot()

        @app.delete("/jobs/{job_id}", status_code=202)
        def cancel_job(job_id: str):
            context = self.cancel(job_id)
            if context is None:
                raise HTTPException(status_code=404, detail="Active job was not found")
            return context._snapshot()

        @app.post("/shutdown", status_code=202)
        def shutdown():
            self.shutdown_event.set()
            return {"state": "stopping"}

        return app

bind(storage_root)

Bind the Server to the Runtime-owned Storage Root.

源代码位于： src\blendjob\server.py

def bind(self, storage_root):
    """Bind the Server to the Runtime-owned Storage Root."""
    storage_root = Path(storage_root).resolve()
    if self.storage_root is not None:
        current = self.storage_root.resolve()
        if current != storage_root:
            raise RuntimeError(
                f"Job Server is already bound to another Storage Root: {current}"
            )
    self.storage_root = storage_root
    self.jobs_directory = storage_root / "jobs"
    for name, factory in tuple(self.resource_factories.items()):
        if name not in self.resources:
            self.resources[name] = factory(self)
    return self

add_resource(name, resource)

Register shared state owned for the full Server lifetime.

源代码位于： src\blendjob\server.py

def add_resource(self, name, resource):
    """Register shared state owned for the full Server lifetime."""
    if not name or "/" in name:
        raise ValueError(f"Invalid resource name: {name}")
    with self.lock:
        if name in self.resources:
            raise ValueError(f"Resource is already registered: {name}")
        self.resources[name] = resource
    return resource

resource(name)

Register a Resource factory initialized after Storage Root binding.

源代码位于： src\blendjob\server.py

def resource(self, name):
    """Register a Resource factory initialized after Storage Root binding."""
    def register(factory):
        if not name or "/" in name:
            raise ValueError(f"Invalid resource name: {name}")
        if name in self.resources or name in self.resource_factories:
            raise ValueError(f"Resource is already registered: {name}")
        self.resource_factories[name] = factory
        if self.storage_root is not None:
            self.resources[name] = factory(self)
        return factory

    return register

clear_resource(name)

Clear an idle Server Resource without racing a new Job submission.

源代码位于： src\blendjob\server.py

def clear_resource(self, name):
    """Clear an idle Server Resource without racing a new Job submission."""
    with self.lock:
        if self.active_job_id is not None or self._queued_job_count():
            return False
        try:
            resource = self.resources[name]
        except KeyError:
            raise KeyError(name) from None
        clear = getattr(resource, "clear", None)
        if clear is None:
            raise TypeError(f"Resource cannot be cleared: {name}")
        clear()
        return True

job(job_type)

Register handler(context, parameters) for a public job type.

源代码位于： src\blendjob\server.py

def job(self, job_type):
    """Register ``handler(context, parameters)`` for a public job type."""
    def register(handler):
        if job_type in self.handlers:
            raise ValueError(f"Job type is already registered: {job_type}")
        self.handlers[job_type] = handler
        return handler

    return register