diff --git a/spec/openapi.yaml b/spec/openapi.yaml index 8a5f4c1..7d38013 100644 --- a/spec/openapi.yaml +++ b/spec/openapi.yaml @@ -1,9 +1,8 @@ -# Comfy API v2 — public spec, vendored into this SDK. +# Comfy API v2 — public specification. # -# GENERATED / VENDORED ONE-WAY — DO NOT HAND-EDIT. +# GENERATED ONE-WAY — DO NOT HAND-EDIT. # Projected automatically from the canonical Comfy API v2 contract and -# synced in by CI. Change the upstream contract, not this copy: the SDK's -# own CI regenerates its low layer from this file and FAILS ON DRIFT. +# synced by CI. Change the upstream contract, not this public copy. openapi: 3.0.3 info: @@ -15,11 +14,12 @@ servers: description: Self-hosted (comfy-api-proxy) - url: https://cloud.comfy.org description: Comfy Cloud -- url: https://{deployment}.comfy.org - description: Serverless deployment (URL shape not final) +- url: https://{deployment}.run.comfy.app + description: Serverless deployment variables: deployment: - default: my-deployment + description: DNS-safe deployment id (subdomain label). Staging uses {deployment}.stg.run.comfy.app. + default: dep-1234abcd-56ef-7890-abcd-ef1234567890 security: - bearerAuth: [] - {} @@ -91,6 +91,12 @@ paths: items: type: string description: Category tags (e.g. `input`). + expires_in: + type: integer + minimum: 60 + maximum: 604800 + description: 'Optional retention override in seconds (60s–7d): the asset''s `expires_at` becomes now + `expires_in`, replacing the platform''s default retention. Implementations without configurable retention ignore it. The bounds apply to this override only — the platform default is operator-configured and may lie outside them.' + example: 86400 responses: '201': description: New blob stored; asset minted. @@ -166,6 +172,12 @@ paths: type: array items: type: string + expires_in: + type: integer + minimum: 60 + maximum: 604800 + description: 'Optional retention override in seconds (60s–7d): the asset''s `expires_at` becomes now + `expires_in`, replacing the platform''s default retention. Implementations without configurable retention ignore it. The bounds apply to this override only — the platform default is operator-configured and may lie outside them.' + example: 86400 responses: '201': description: Asset minted over the existing blob. @@ -233,6 +245,44 @@ paths: $ref: '#/components/responses/NotFound' '500': $ref: '#/components/responses/UpstreamError' + delete: + operationId: deleteAsset + tags: + - assets + summary: Delete an asset record + description: 'Deletes the asset RECORD. The underlying content-addressed blob is + + untouched while any other asset still references it (hash dedup means + + blobs are shared) — deleting an asset never destroys another asset''s + + bytes. + + + A second delete of the same id returns `404`, indistinguishable from + + an id that never existed or belongs to another account. + + ' + parameters: + - $ref: '#/components/parameters/AssetId' + responses: + '204': + description: Record deleted. + '401': + $ref: '#/components/responses/Unauthorized' + '403': + $ref: '#/components/responses/Forbidden' + '404': + $ref: '#/components/responses/NotFound' + '409': + description: '`asset_in_use` — the record cannot be deleted while the platform still depends on it. Each surface defines its own holds (for example: a job''s outputs reference the record, or a content-moderation workflow requires it to be preserved); the response body deliberately never says which hold applies.' + content: + application/json: + schema: + $ref: '#/components/schemas/ErrorEnvelope' + '500': + $ref: '#/components/responses/UpstreamError' /api/v2/assets/{id}/content: get: operationId: getAssetContent @@ -394,7 +444,7 @@ paths: schema: $ref: '#/components/schemas/ErrorEnvelope' '429': - description: '`queue_full` — bounded queue depth reached.' + description: '`queue_full` (bounded queue depth reached) or, on deployment-scoped surfaces, `deployment_not_ready` (deployment still provisioning/starting). Disambiguate by `error.code`; both mean back off and retry after `Retry-After`.' headers: Retry-After: $ref: '#/components/headers/RetryAfter' @@ -440,6 +490,32 @@ paths: $ref: '#/components/responses/RateLimited' '500': $ref: '#/components/responses/UpstreamError' + /api/v2/jobs/{id}/workflow: + get: + operationId: getJobWorkflow + tags: + - jobs + summary: The workflow behind a job — authoring version if pinned, executed graph otherwise + description: "Returns the workflow behind a job. The response's `format` field says\nwhich of two different shapes `workflow` is in:\n\n- `format: save` — the original authoring workflow exactly as saved\n in the Comfy Cloud editor at the version the job ran, including\n canvas layout and frontend-only nodes (e.g. Note nodes; Get/Set\n nodes not yet expanded). Returned only when the job is pinned to a\n specific workflow version — see the \"when you get which\" note\n below.\n- `format: api` — the executed API-format prompt graph the job\n actually ran: frontend-only constructs are gone and Get/Set nodes\n are expanded. This is the same shape `POST /api/v2/jobs`'s\n `workflow` request field takes, and never includes the\n submission's `extra_data`, which can carry a live credential.\n\nAlways branch on `format`, never assume one or the other — which\nshape comes back depends on how the job was submitted, not on\nanything the caller controls per-request.\n\nA deliberate sub-resource, not a field on `GET /api/v2/jobs/{id}` —\nso the polling workhorse stays cheap and a caller pays for this only\nwhen it actually wants the workflow (for example, to recover what\nproduced a given output).\n\nTied to the job's own retention: this 404s under the same conditions\n`GET /api/v2/jobs/{id}` does (unknown, not-yours, or past its\nretention deadline) — there is no separate lifetime for the\nworkflow.\n\n**When you get which:** a job only carries a pinned workflow version\nwhen it was submitted with that association. Today that means jobs\nsubmitted from the Comfy Cloud frontend/editor. Jobs submitted\ndirectly through this v2 API (`POST /api/v2/jobs`) do not carry that\nassociation — v2 job submission has no version-linking fields yet —\nso they always get `format: api`. This is expected, not a bug: it\nwill change once v2 submission grows the same version pinning.\n\nA job pinned to a version also falls back to `format: api` if that\nversion, or the workflow it belongs to, is no longer readable by the\ncaller — for example the caller deleted the workflow since the job\nran. This is the same fallback as an unpinned job, and for the same\nreason: it is preferable to the alternative of erroring the whole\nrequest over data that is genuinely gone.\n" + parameters: + - $ref: '#/components/parameters/JobId' + responses: + '200': + description: The workflow graph. + content: + application/json: + schema: + $ref: '#/components/schemas/JobWorkflowResponse' + '401': + $ref: '#/components/responses/Unauthorized' + '403': + $ref: '#/components/responses/Forbidden' + '404': + $ref: '#/components/responses/NotFound' + '429': + $ref: '#/components/responses/RateLimited' + '500': + $ref: '#/components/responses/UpstreamError' /api/v2/jobs/{id}/events: get: operationId: getJobEvents @@ -573,14 +649,14 @@ components: required: true schema: type: string - example: job_01JZTGXW9Q2M4R8V0B1N3P5D7F + example: 7f3d2c1b-9a8e-4d6f-b012-3c4d5e6f7a8b AssetId: name: id in: path required: true schema: type: string - example: asset_01JZV8Q3M7K2W9X0Y1Z2A3B4C5 + example: 9f8a1c0d-2b3e-4f56-8a7b-1c2d3e4f5a6b BlakeHash: name: hash in: path @@ -643,7 +719,7 @@ components: properties: id: type: string - example: asset_01JZV8Q3M7K2W9X0Y1Z2A3B4C5 + example: 9f8a1c0d-2b3e-4f56-8a7b-1c2d3e4f5a6b hash: type: string nullable: true @@ -673,6 +749,15 @@ components: url_expires_at: type: string format: date-time + expires_at: + type: string + format: date-time + nullable: true + description: 'Retention deadline for the asset itself (distinct from `url_expires_at`, the signed URL''s validity). Null or absent means the asset is non-expiring. On a dedup-hit create response the deadline may be later than now + the requested/default retention: re-referencing content extends its retention, never shortens it.' + job_id: + type: string + nullable: true + description: ID of the job that produced this asset. Absent for uploaded assets, which have no producing job. Job: type: object description: One execution of a workflow. Durable from creation until `expires_at`; `outputs` populates incrementally during execution. @@ -691,7 +776,7 @@ components: properties: id: type: string - example: job_01JZTGXW9Q2M4R8V0B1N3P5D7F + example: 7f3d2c1b-9a8e-4d6f-b012-3c4d5e6f7a8b status: $ref: '#/components/schemas/JobStatus' created_at: @@ -736,6 +821,23 @@ components: execution_ms: 42000 urls: $ref: '#/components/schemas/JobUrls' + JobWorkflowResponse: + type: object + description: The workflow behind a job. See GET /api/v2/jobs/{id}/workflow's description for exactly when `format` is `save` vs `api`. + required: + - workflow + - format + properties: + workflow: + type: object + description: The workflow, verbatim, in the shape `format` says. + additionalProperties: true + format: + type: string + enum: + - save + - api + description: 'Discriminates the `workflow` field''s shape. `save`: the original authoring workflow JSON, at the version pinned to the job. `api`: the executed API-format prompt graph.' JobStatus: type: string enum: @@ -755,7 +857,7 @@ components: ' JobUrls: type: object - description: Embedded follow-up links — follow these, don't build URLs. + description: Embedded follow-up links — follow these, don't build URLs. A link is either an absolute URL or a host-relative reference (leading `/`) that already includes any prefix the serving surface is mounted under (e.g. a serverless gateway's `/deployment/{deployment_id}/api/v2`). Clients MUST resolve a host-relative link against the request origin (scheme + authority), never against a configured base URL — joining it to a base URL that carries the same mount prefix duplicates the prefix. required: - self - events @@ -843,7 +945,7 @@ components: id: type: string description: Asset UUID. - example: asset_01JZV9R4N8... + example: 9f8a1c0d-2b3e-4f56-... hash: type: string nullable: true @@ -854,6 +956,10 @@ components: url_expires_at: type: string format: date-time + job_id: + type: string + nullable: true + description: ID of the job that produced this output. OutputType: type: string enum: @@ -899,6 +1005,18 @@ components: `not_found` (404), `unauthorized` (401), `forbidden` (403). + Deployment-scoped surfaces add: `deployment_not_ready` (429 + + + Retry-After — the deployment can still reach ready; retry) and + + `deployment_stopped` (422 — terminal deployment state; a retry + + cannot succeed without operator action). A 429 is disambiguated + + by `error.code` alone; clients should treat any 429 + Retry-After + + as "back off and retry". + ' required: - error @@ -964,7 +1082,7 @@ components: type: string AssetReference: type: object - description: "The typed asset-reference object placed inside workflow JSON where a\nfilename would normally go (documented here for tooling; it is not a\nrequest/response body itself):\n\n {\"__type\": \"core/ASSET\",\n \"info\": {\"id\": \"asset_...\", \"hash\": \"blake3:...\",\n \"file_path\": \"photo.png\"}}\n\n`info.id` (the asset UUID) is required in v1 and authoritative;\n`hash` and `file_path` are optional staging/lookup hints and never\noverride a present `id`. A malformed reference or one that is not\nresolvable/owned by the caller fails submission with 422\n`missing_asset`.\n" + description: "The typed asset-reference object placed inside workflow JSON where a\nfilename would normally go (documented here for tooling; it is not a\nrequest/response body itself):\n\n {\"__type\": \"core/ASSET\",\n \"info\": {\"id\": \"\", \"hash\": \"blake3:...\",\n \"file_path\": \"photo.png\"}}\n\n`info.id` (the asset UUID) is required in v1 and authoritative;\n`hash` and `file_path` are optional staging/lookup hints and never\noverride a present `id`. A malformed reference or one that is not\nresolvable/owned by the caller fails submission with 422\n`missing_asset`.\n" required: - __type - info @@ -980,7 +1098,7 @@ components: properties: id: type: string - example: asset_01JZV8Q3M7K2W9X0Y1Z2A3B4C5 + example: 9f8a1c0d-2b3e-4f56-8a7b-1c2d3e4f5a6b hash: type: string example: blake3:9f8a1c0d... diff --git a/src/comfy_low/__init__.py b/src/comfy_low/__init__.py index 88da7d7..75d65d8 100644 --- a/src/comfy_low/__init__.py +++ b/src/comfy_low/__init__.py @@ -41,9 +41,11 @@ "assetFromHash", "headAssetByHash", "getAsset", + "deleteAsset", "getAssetContent", "postJobs", "getJob", + "getJobWorkflow", "getJobEvents", "cancelJob", } @@ -55,9 +57,11 @@ "assetFromHash": "asset_from_hash", "headAssetByHash": "head_asset_by_hash", "getAsset": "get_asset", + "deleteAsset": "delete_asset", "getAssetContent": "get_asset_content", "postJobs": "post_jobs", "getJob": "get_job", + "getJobWorkflow": "get_job_workflow", "getJobEvents": "get_job_events", "cancelJob": "cancel_job", } diff --git a/src/comfy_low/models/__init__.py b/src/comfy_low/models/__init__.py index 8defec6..e8b3b86 100644 --- a/src/comfy_low/models/__init__.py +++ b/src/comfy_low/models/__init__.py @@ -17,10 +17,12 @@ AssetReference, Error, ErrorEnvelope, + Format, Job, JobError, JobStatus, JobUrls, + JobWorkflowResponse, LogEvent, Output, OutputType, @@ -34,10 +36,12 @@ "AssetReference", "Error", "ErrorEnvelope", + "Format", "Job", "JobError", "JobStatus", "JobUrls", + "JobWorkflowResponse", "LogEvent", "Output", "OutputType", diff --git a/src/comfy_low/models/_generated.py b/src/comfy_low/models/_generated.py index d33fe69..c188060 100644 --- a/src/comfy_low/models/_generated.py +++ b/src/comfy_low/models/_generated.py @@ -13,7 +13,7 @@ class Asset(BaseModel): A user-owned record identified by a server-assigned UUID, backing an immutable blob whose content carries a server-computed blake3 hash. `hash` may be computed lazily: an asset record (and its retrievable bytes) can exist before its hash is filled in. """ - id: Annotated[str, Field(examples=['asset_01JZV8Q3M7K2W9X0Y1Z2A3B4C5'])] + id: Annotated[str, Field(examples=['9f8a1c0d-2b3e-4f56-8a7b-1c2d3e4f5a6b'])] hash: Annotated[ str | None, Field( @@ -35,6 +35,44 @@ class Asset(BaseModel): AnyUrl, Field(description='Short-lived content URL (signed, or proxy-served).') ] url_expires_at: AwareDatetime + expires_at: Annotated[ + AwareDatetime | None, + Field( + description="Retention deadline for the asset itself (distinct from `url_expires_at`, the signed URL's validity). Null or absent means the asset is non-expiring. On a dedup-hit create response the deadline may be later than now + the requested/default retention: re-referencing content extends its retention, never shortens it." + ), + ] = None + job_id: Annotated[ + str | None, + Field( + description='ID of the job that produced this asset. Absent for uploaded assets, which have no producing job.' + ), + ] = None + + +class Format(Enum): + """ + Discriminates the `workflow` field's shape. `save`: the original authoring workflow JSON, at the version pinned to the job. `api`: the executed API-format prompt graph. + """ + + save = 'save' + api = 'api' + + +class JobWorkflowResponse(BaseModel): + """ + The workflow behind a job. See GET /api/v2/jobs/{id}/workflow's description for exactly when `format` is `save` vs `api`. + """ + + workflow: Annotated[ + dict[str, Any], + Field(description='The workflow, verbatim, in the shape `format` says.'), + ] + format: Annotated[ + Format, + Field( + description="Discriminates the `workflow` field's shape. `save`: the original authoring workflow JSON, at the version pinned to the job. `api`: the executed API-format prompt graph." + ), + ] class JobStatus(Enum): @@ -56,7 +94,7 @@ class JobStatus(Enum): class JobUrls(BaseModel): """ - Embedded follow-up links — follow these, don't build URLs. + Embedded follow-up links — follow these, don't build URLs. A link is either an absolute URL or a host-relative reference (leading `/`) that already includes any prefix the serving surface is mounted under (e.g. a serverless gateway's `/deployment/{deployment_id}/api/v2`). Clients MUST resolve a host-relative link against the request origin (scheme + authority), never against a configured base URL — joining it to a base URL that carries the same mount prefix duplicates the prefix. """ self: str @@ -136,6 +174,12 @@ class ErrorEnvelope(BaseModel): (404), `idempotency_key_reuse` (422), `queue_full` (429 + Retry-After), `insufficient_credits` (402), `not_found` (404), `unauthorized` (401), `forbidden` (403). + Deployment-scoped surfaces add: `deployment_not_ready` (429 + + Retry-After — the deployment can still reach ready; retry) and + `deployment_stopped` (422 — terminal deployment state; a retry + cannot succeed without operator action). A 429 is disambiguated + by `error.code` alone; clients should treat any 429 + Retry-After + as "back off and retry". """ @@ -175,7 +219,7 @@ class FieldType(Enum): class Info(BaseModel): - id: Annotated[str, Field(examples=['asset_01JZV8Q3M7K2W9X0Y1Z2A3B4C5'])] + id: Annotated[str, Field(examples=['9f8a1c0d-2b3e-4f56-8a7b-1c2d3e4f5a6b'])] hash: Annotated[str | None, Field(examples=['blake3:9f8a1c0d...'])] = None file_path: Annotated[str | None, Field(examples=['photo.png'])] = None @@ -187,7 +231,7 @@ class AssetReference(BaseModel): request/response body itself): {"__type": "core/ASSET", - "info": {"id": "asset_...", "hash": "blake3:...", + "info": {"id": "", "hash": "blake3:...", "file_path": "photo.png"}} `info.id` (the asset UUID) is required in v1 and authoritative; @@ -213,13 +257,16 @@ class Output(BaseModel): content_type: Annotated[str, Field(examples=['image/png'])] size_bytes: Annotated[int, Field(examples=[1848320])] id: Annotated[ - str, Field(description='Asset UUID.', examples=['asset_01JZV9R4N8...']) + str, Field(description='Asset UUID.', examples=['9f8a1c0d-2b3e-4f56-...']) ] hash: Annotated[ str | None, Field(description='`blake3:`; null until lazily computed.') ] url: AnyUrl url_expires_at: AwareDatetime + job_id: Annotated[ + str | None, Field(description='ID of the job that produced this output.') + ] = None class Job(BaseModel): @@ -227,7 +274,7 @@ class Job(BaseModel): One execution of a workflow. Durable from creation until `expires_at`; `outputs` populates incrementally during execution. """ - id: Annotated[str, Field(examples=['job_01JZTGXW9Q2M4R8V0B1N3P5D7F'])] + id: Annotated[str, Field(examples=['7f3d2c1b-9a8e-4d6f-b012-3c4d5e6f7a8b'])] status: JobStatus created_at: AwareDatetime started_at: Annotated[AwareDatetime | None, Field(...)] diff --git a/src/comfy_low/transport.py b/src/comfy_low/transport.py index 207f6bf..e8f84fb 100644 --- a/src/comfy_low/transport.py +++ b/src/comfy_low/transport.py @@ -32,7 +32,7 @@ from . import _multipart from .errors import ApiError, error_from_envelope -from .models import Asset, Job +from .models import Asset, Job, JobWorkflowResponse from .sse import RawEvent, SSEDecoder _API = "/api/v2" @@ -341,6 +341,11 @@ def get_asset(self, asset_id: str, *, timeout: Any = _UNSET) -> Asset: resp = self.raw_request("GET", f"/assets/{asset_id}", timeout=timeout) return Asset.model_validate(self._p.parse_or_raise(resp, (200,))) + def delete_asset(self, asset_id: str, *, timeout: Any = _UNSET) -> None: + """DELETE /api/v2/assets/{id} — removes the asset record and its content.""" + resp = self.raw_request("DELETE", f"/assets/{asset_id}", timeout=timeout) + self._p.parse_or_raise(resp, (204,)) + @contextmanager def get_asset_content( self, @@ -462,6 +467,14 @@ def cancel_job(self, job_id_or_url: str, *, timeout: Any = _UNSET) -> Job: resp = self.raw_request("POST", path, timeout=timeout) return Job.model_validate(self._p.parse_or_raise(resp, (200,))) + def get_job_workflow(self, job_id_or_url: str, *, timeout: Any = _UNSET) -> JobWorkflowResponse: + """GET /api/v2/jobs/{id}/workflow — the workflow graph behind a job.""" + path = ( + job_id_or_url if _looks_like_path(job_id_or_url) else f"/jobs/{job_id_or_url}/workflow" + ) + resp = self.raw_request("GET", path, timeout=timeout) + return JobWorkflowResponse.model_validate(self._p.parse_or_raise(resp, (200,))) + class AsyncComfyLow: """Asynchronous protocol bindings — mirrors :class:`ComfyLow`.""" @@ -617,6 +630,11 @@ async def get_asset(self, asset_id: str, *, timeout: Any = _UNSET) -> Asset: resp = await self.raw_request("GET", f"/assets/{asset_id}", timeout=timeout) return Asset.model_validate(self._p.parse_or_raise(resp, (200,))) + async def delete_asset(self, asset_id: str, *, timeout: Any = _UNSET) -> None: + """DELETE /api/v2/assets/{id} — removes the asset record and its content.""" + resp = await self.raw_request("DELETE", f"/assets/{asset_id}", timeout=timeout) + self._p.parse_or_raise(resp, (204,)) + @asynccontextmanager async def get_asset_content( self, @@ -709,6 +727,16 @@ async def cancel_job(self, job_id_or_url: str, *, timeout: Any = _UNSET) -> Job: resp = await self.raw_request("POST", path, timeout=timeout) return Job.model_validate(self._p.parse_or_raise(resp, (200,))) + async def get_job_workflow( + self, job_id_or_url: str, *, timeout: Any = _UNSET + ) -> JobWorkflowResponse: + """Async :meth:`ComfyLow.get_job_workflow`.""" + path = ( + job_id_or_url if _looks_like_path(job_id_or_url) else f"/jobs/{job_id_or_url}/workflow" + ) + resp = await self.raw_request("GET", path, timeout=timeout) + return JobWorkflowResponse.model_validate(self._p.parse_or_raise(resp, (200,))) + def _looks_like_path(s: str) -> bool: return s.startswith("http") or s.startswith("/") diff --git a/src/comfy_sdk/__init__.py b/src/comfy_sdk/__init__.py index 399fe99..69fbd51 100644 --- a/src/comfy_sdk/__init__.py +++ b/src/comfy_sdk/__init__.py @@ -55,7 +55,7 @@ Unauthorized, WorkflowFormatUi, ) -from .jobs import AsyncJob, Job +from .jobs import AsyncJob, Job, JobWorkflow from .outputs import AsyncOutput, DownloadUrl, Output from .workflows import Workflow, WorkflowFactory @@ -82,6 +82,7 @@ "WorkflowFactory", "Job", "AsyncJob", + "JobWorkflow", "Output", "AsyncOutput", "DownloadUrl", diff --git a/src/comfy_sdk/assets.py b/src/comfy_sdk/assets.py index 9ff9c4b..23d092d 100644 --- a/src/comfy_sdk/assets.py +++ b/src/comfy_sdk/assets.py @@ -128,6 +128,14 @@ def commit(self) -> str: assert self._id is not None return self._id + def delete(self) -> None: + """Delete this asset from storage.""" + if self._id is None: + raise RuntimeError("cannot delete an uncommitted asset") + with translating(): + self._low.delete_asset(self._id) + self._id = None + def as_reference(self) -> dict[str, object]: """The ``core/ASSET`` object (commits first if needed).""" self.commit() @@ -166,6 +174,14 @@ async def commit(self) -> str: assert self._id is not None return self._id + async def delete(self) -> None: + """Delete this asset from storage.""" + if self._id is None: + raise RuntimeError("cannot delete an uncommitted asset") + with translating(): + await self._low.delete_asset(self._id) + self._id = None + async def as_reference(self) -> dict[str, object]: await self.commit() assert self._id is not None @@ -242,6 +258,11 @@ def get(self, asset_id: str) -> Asset: asset._apply(model) return asset + def delete(self, asset_id: str) -> None: + """Delete an asset by UUID.""" + with translating(): + self._low.delete_asset(asset_id) + class AsyncAssetFactory: """``client.assets`` — async alternative constructors for :class:`AsyncAsset`.""" @@ -286,6 +307,11 @@ async def get(self, asset_id: str) -> AsyncAsset: asset._apply(model) return asset + async def delete(self, asset_id: str) -> None: + """Delete an asset by UUID.""" + with translating(): + await self._low.delete_asset(asset_id) + def _no_opener() -> tuple[BinaryIO, int | None]: raise RuntimeError("this asset is already committed; nothing to upload") diff --git a/src/comfy_sdk/jobs.py b/src/comfy_sdk/jobs.py index 59872d9..0e18a58 100644 --- a/src/comfy_sdk/jobs.py +++ b/src/comfy_sdk/jobs.py @@ -12,7 +12,8 @@ import time from collections.abc import AsyncIterator, Iterator -from typing import Any +from dataclasses import dataclass +from typing import Any, Literal import httpx @@ -29,6 +30,24 @@ _RECONNECT_PAUSE = 0.1 +@dataclass(frozen=True) +class JobWorkflow: + """The workflow that produced a job — see :meth:`Job.get_workflow`. + + ``format`` discriminates the shape of ``graph``, so a caller can tell + which one it holds instead of the two shapes being silently collapsed: + + * ``"api"`` — the executed graph. API-format, so frontend-only constructs + (Note nodes, Get/Set) are already resolved away. + * ``"save"`` — the authoring workflow at the version the job ran, + un-mangled. Only occurs for a job that pins a workflow version; a job + submitted through this SDK today always gets ``"api"``. + """ + + graph: dict[str, Any] + format: Literal["save", "api"] + + class Job: """Synchronous job handle.""" @@ -108,6 +127,19 @@ def cancel(self) -> Job: self._model = self._low.cancel_job(self._model.urls.cancel or self._model.id) return self + def get_workflow(self) -> JobWorkflow: + """Fetch the workflow that produced this job. + + Needed for a job rehydrated purely by id (e.g. via + ``client.jobs.get``) — the SDK only holds the workflow it submitted + for as long as the same process's :class:`Job` handle is alive, so + this is the only way to see the graph otherwise. A missing job raises + the SDK's normal :class:`~comfy_sdk.exceptions.NotFound`. + """ + with translating(): + data = self._low.get_job_workflow(self._model.id) + return JobWorkflow(graph=data.workflow, format=data.format.value) + # -- live events (best-effort, reconnecting) -------------------------- def events(self) -> Iterator[Event]: """Typed live event iterator. Auto-reconnects with no replay; falls back @@ -217,6 +249,12 @@ async def cancel(self) -> AsyncJob: self._model = await self._low.cancel_job(self._model.urls.cancel or self._model.id) return self + async def get_workflow(self) -> JobWorkflow: + """Async :meth:`Job.get_workflow`.""" + with translating(): + data = await self._low.get_job_workflow(self._model.id) + return JobWorkflow(graph=data.workflow, format=data.format.value) + async def events(self) -> AsyncIterator[Event]: """Async :meth:`Job.events` — typed live stream, auto-reconnecting with no replay and the poll path as its backstop. diff --git a/tests/conftest.py b/tests/conftest.py index 4f547e8..072bdd2 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -55,11 +55,20 @@ class ServerState: # Set to a list of output dicts (see `_output_json`) to test a job with # multiple outputs, each backed by a distinct asset id. job_outputs: list[dict] | None = None + # GET /jobs/{id}/workflow response. `job_workflow_not_found=True` answers + # 404 job_not_found instead, for the missing-job path. + job_workflow_graph: dict[str, Any] = field( + default_factory=lambda: {"3": {"class_type": "KSampler", "inputs": {}}} + ) + job_workflow_format: str = "api" + job_workflow_not_found: bool = False # --- counters the tests assert on --- upload_count: int = 0 from_hash_count: int = 0 head_count: int = 0 + delete_count: int = 0 + deleted_assets: set[str] = field(default_factory=set) job_poll_count: int = 0 events_connect_count: int = 0 submit_count: int = 0 @@ -176,6 +185,20 @@ def do_HEAD(self) -> None: self.send_response(404) self.end_headers() + # -- DELETE -- + def do_DELETE(self) -> None: + if not self._auth_ok(): + self._err(401, "unauthorized", "no key") + return + m = re.match(r"/api/v2/assets/([^/]+)$", self.path) + if m: + state.delete_count += 1 + state.deleted_assets.add(m.group(1)) + self.send_response(204) + self.end_headers() + return + self._err(404, "not_found") + # -- GET -- def do_GET(self) -> None: if not self._auth_ok(): @@ -184,6 +207,9 @@ def do_GET(self) -> None: m = re.match(r"/api/v2/assets/([^/]+)/content$", self.path) if m: + if m.group(1) in state.deleted_assets: + self._err(404, "not_found") + return if state.redirect_content_to: self._redirect(state.redirect_content_to) else: @@ -191,12 +217,19 @@ def do_GET(self) -> None: return m = re.match(r"/api/v2/assets/([^/]+)$", self.path) if m: + if m.group(1) in state.deleted_assets: + self._err(404, "not_found") + return self._json(200, _asset_json(m.group(1), state.server_hash, False, 33)) return m = re.match(r"/api/v2/jobs/([^/]+)/events$", self.path) if m: self._serve_events(m.group(1)) return + m = re.match(r"/api/v2/jobs/([^/]+)/workflow$", self.path) + if m: + self._serve_job_workflow(m.group(1)) + return m = re.match(r"/api/v2/jobs/([^/]+)$", self.path) if m: self._serve_job(m.group(1)) @@ -243,6 +276,15 @@ def _serve_job(self, job_id: str) -> None: outputs = [] self._json(200, _job_json(job_id, status, outputs)) + def _serve_job_workflow(self, job_id: str) -> None: + if state.job_workflow_not_found: + self._err(404, "job_not_found", "no such job") + return + self._json( + 200, + {"workflow": state.job_workflow_graph, "format": state.job_workflow_format}, + ) + def _serve_events(self, job_id: str) -> None: state.events_connect_count += 1 if state.events_not_implemented: diff --git a/tests/test_assets.py b/tests/test_assets.py index aad89b1..d90946b 100644 --- a/tests/test_assets.py +++ b/tests/test_assets.py @@ -6,6 +6,7 @@ import pytest +from comfy_low.errors import NotFound from comfy_sdk import Comfy, HashMismatch @@ -102,3 +103,36 @@ def test_hash_mismatch_surfaced_without_blind_retry(server, tmp_path) -> None: # Exactly one upload attempt — a 409 hash_mismatch must not be blindly retried. assert server.state.upload_count == 1 + + +def test_delete_asset_by_id(server) -> None: + with Comfy() as client: + client.assets.delete("asset_uuid_01") + with pytest.raises(NotFound): + client.assets.get("asset_uuid_01") + + assert server.state.delete_count == 1 + + +def test_delete_asset_on_asset_instance(server) -> None: + data = b"delete-me-bytes" + with Comfy() as client: + asset = client.assets.from_bytes(data, filename="photo.png") + asset.commit() + asset_id = asset.id + asset.delete() + with pytest.raises(NotFound): + client.assets.get(asset_id) + + assert asset_id == "asset_uploaded_01" + assert server.state.delete_count == 1 + assert asset.id is None + + +def test_delete_uncommitted_asset_raises(server) -> None: + with Comfy() as client: + asset = client.assets.from_bytes(b"not-uploaded", filename="photo.png") + with pytest.raises(RuntimeError, match="uncommitted"): + asset.delete() + + assert server.state.delete_count == 0 diff --git a/tests/test_async.py b/tests/test_async.py index a9023dd..f630fd7 100644 --- a/tests/test_async.py +++ b/tests/test_async.py @@ -4,6 +4,7 @@ import pytest +from comfy_low.errors import NotFound from comfy_sdk import AsyncComfy, MissingAsset, Progress, StatusChange @@ -147,3 +148,35 @@ async def test_async_queue_full_retries_with_retry_after(server) -> None: async with AsyncComfy() as client: await client.submit(_wf(client)) assert server.state.submit_count == 3 + + +async def test_async_delete_asset_by_id(server) -> None: + async with AsyncComfy() as client: + await client.assets.delete("asset_uuid_01") + with pytest.raises(NotFound): + await client.assets.get("asset_uuid_01") + + assert server.state.delete_count == 1 + + +async def test_async_delete_asset_on_asset_instance(server) -> None: + data = b"async-delete-me-bytes" + async with AsyncComfy() as client: + asset = client.assets.from_bytes(data, filename="photo.png") + asset_id = await asset.commit() + await asset.delete() + with pytest.raises(NotFound): + await client.assets.get(asset_id) + + assert asset_id == "asset_uploaded_01" + assert server.state.delete_count == 1 + assert asset.id is None + + +async def test_async_delete_uncommitted_asset_raises(server) -> None: + async with AsyncComfy() as client: + asset = client.assets.from_bytes(b"not-uploaded", filename="photo.png") + with pytest.raises(RuntimeError, match="uncommitted"): + await asset.delete() + + assert server.state.delete_count == 0 diff --git a/tests/test_job_workflow.py b/tests/test_job_workflow.py new file mode 100644 index 0000000..80b1992 --- /dev/null +++ b/tests/test_job_workflow.py @@ -0,0 +1,78 @@ +"""Job.get_workflow() / AsyncJob.get_workflow() — GET /api/v2/jobs/{id}/workflow. + +The stub server in conftest.py stands in for the real endpoint. +""" + +from __future__ import annotations + +import pytest + +from comfy_sdk import AsyncComfy, Comfy, JobWorkflow, NotFound + + +def _wf(client: Comfy | AsyncComfy): + return client.workflows.from_json({"3": {"class_type": "KSampler", "inputs": {}}}) + + +def test_get_workflow_returns_api_format(server) -> None: + server.state.job_workflow_format = "api" + server.state.job_workflow_graph = {"3": {"class_type": "KSampler", "inputs": {}}} + with Comfy() as client: + job = client.submit(_wf(client)) + wf = job.get_workflow() + assert isinstance(wf, JobWorkflow) + assert wf.format == "api" + assert wf.graph == {"3": {"class_type": "KSampler", "inputs": {}}} + + +def test_get_workflow_returns_save_format(server) -> None: + # "save" is the authoring workflow at the pinned version, un-mangled — + # a different shape than the executed "api" graph, so both must be + # reachable, not collapsed into one. + server.state.job_workflow_format = "save" + server.state.job_workflow_graph = { + "nodes": [{"id": 3, "type": "KSampler"}], + "links": [], + "last_node_id": 3, + } + with Comfy() as client: + job = client.submit(_wf(client)) + wf = job.get_workflow() + assert wf.format == "save" + assert wf.graph["nodes"][0]["type"] == "KSampler" + + +def test_get_workflow_on_job_rehydrated_by_id(server) -> None: + # The motivating case: a job the SDK did not submit in this process (e.g. + # rehydrated purely by id via client.jobs.get) still exposes its workflow. + with Comfy() as client: + submitted = client.submit(_wf(client)) + rehydrated = client.jobs.get(submitted.id) + wf = rehydrated.get_workflow() + assert wf.format == "api" + + +def test_get_workflow_not_found_raises_sdk_not_found(server) -> None: + server.state.job_workflow_not_found = True + with Comfy() as client: + job = client.submit(_wf(client)) + with pytest.raises(NotFound): + job.get_workflow() + + +async def test_async_get_workflow_mirrors_sync(server) -> None: + server.state.job_workflow_format = "save" + server.state.job_workflow_graph = {"nodes": [], "links": [], "last_node_id": 0} + async with AsyncComfy() as client: + job = await client.submit(_wf(client)) + wf = await job.get_workflow() + assert wf.format == "save" + assert wf.graph == {"nodes": [], "links": [], "last_node_id": 0} + + +async def test_async_get_workflow_not_found_raises_sdk_not_found(server) -> None: + server.state.job_workflow_not_found = True + async with AsyncComfy() as client: + job = await client.submit(_wf(client)) + with pytest.raises(NotFound): + await job.get_workflow()