magick_mind.resources.v1.artifact

Artifact resource for Magick Mind SDK v1 API.

Provides methods for file upload via presigned S3 URLs and artifact management.

ArtifactResourceV1

class ArtifactResourceV1(BaseResource)

Artifact resource for managing file uploads and artifacts.

Implements presigned S3 upload flow:

1. Call presign_upload() to get a presigned URL
2. Upload file directly to S3 using the presigned URL
3. Webhook or finalize() confirms completion
4. Use get() or list() to retrieve artifact metadata

ArtifactResourceV1.presign_upload

async def presign_upload(
        file_name: str,
        content_type: str,
        size_bytes: int,
        end_user_id: Optional[str] = None,
        corpus_id: Optional[str] = None) -> PresignArtifactResponse

Get a presigned S3 URL for uploading a file.

Arguments:

• file_name - Name of the file to upload (required)
• content_type - MIME type of the file (e.g., 'application/pdf', 'image/png')
• size_bytes - File size in bytes (must be > 0)
• end_user_id - Optional end user identifier
• corpus_id - Optional corpus ID to associate artifact with

Returns:

PresignArtifactResponse containing upload_url and metadata

Example:

Get presigned URL

response = await client.v1.artifact.presign_upload( file_name="document.pdf", content_type="application/pdf", size_bytes=1024000, corpus_id="corpus-123" )

Upload file to S3 using httpx

import httpx with open("document.pdf", "rb") as f: httpx.put( response.upload_url, content=f, headers=response.required_headers )

print(f"Artifact ID: {response.id}")

ArtifactResourceV1.upload_file

async def upload_file(
    file_path: str,
    content_type: str,
    end_user_id: Optional[str] = None,
    corpus_id: Optional[str] = None
) -> tuple[PresignArtifactResponse, "Response"]

Convenience method to presign and upload a file in one call.

This is a high-level wrapper that:

1. Gets the file size
2. Obtains presigned URL
3. Uploads the file to S3

Arguments:

• file_path - Path to the file to upload
• content_type - MIME type of the file
• end_user_id - Optional end user identifier
• corpus_id - Optional corpus ID to associate artifact with

Returns:

Tuple of (PresignArtifactResponse, S3 upload response)

Example:

presign_resp, upload_resp = await client.v1.artifact.upload_file( file_path="./document.pdf", content_type="application/pdf", corpus_id="corpus-123" )

if upload_resp.status_code == 200: print(f"Upload successful! Artifact ID: {presign_resp.id}")

ArtifactResourceV1.get

async def get(artifact_id: str) -> Artifact

Get artifact metadata by ID.

Arguments:

• artifact_id - The artifact ID to retrieve

Returns:

Artifact object with metadata

Example:

artifact = await client.v1.artifact.get(artifact_id="art-123")

• print(f"Status - {artifact.status}") print(f"S3 URL: {artifact.s3_url}")

ArtifactResourceV1.list

async def list(end_user_id: Optional[str] = None,
               status: Optional[str] = None,
               cursor: Optional[str] = None,
               limit: Optional[int] = None) -> list[Artifact]

List artifacts for the authenticated tenant with optional filters.

Arguments:

• end_user_id - Filter by end user ID (optional)
• status - Filter by status — uploaded, processing, ready, failed, deleted
• cursor - Pagination cursor (opaque string from a previous response)
• limit - Maximum number of results to return

Returns:

List of Artifact objects

Example:

Get all artifacts

artifacts = await client.v1.artifact.list() for artifact in artifacts: print(f"- {artifact.id}: {artifact.status}")

Get ready artifacts with pagination

ready = await client.v1.artifact.list(status="ready", limit=20)

ArtifactResourceV1.delete

async def delete(artifact_id: str) -> None

Delete an artifact.

Raises ProblemDetailsException with status 404 if the artifact is not found, or status 410 if it has already been deleted.

Arguments:

• artifact_id - The artifact ID to delete

Example:

await client.v1.artifact.delete(artifact_id="art-123") print("Artifact deleted successfully")

ArtifactResourceV1.download_url

async def download_url(artifact_id: str) -> DownloadUrlResponse

Get a presigned download URL for an artifact.

The URL is short-lived — check expires_at before caching.

Arguments:

• artifact_id - The artifact ID

Returns:

DownloadUrlResponse with download_url, expires_at, and file_name fields.

Example:

dl = await client.v1.artifact.download_url("art-123")

• print(f"Download - {dl.download_url}")

ArtifactResourceV1.finalize

async def finalize(
        artifact_id: str,
        bucket: str,
        key: str,
        version_id: Optional[str] = None,
        size_bytes: Optional[int] = None,
        content_type: Optional[str] = None,
        etag: Optional[str] = None,
        checksum_sha256: Optional[str] = None,
        corpus_id: Optional[str] = None) -> FinalizeArtifactResponse

Client-driven finalize (fallback when webhook is unavailable).

This is typically used in local development or when the S3 Lambda webhook path is not available. In production, webhooks handle finalization automatically.

Arguments:

• artifact_id - Artifact ID from presign response
• bucket - S3 bucket name
• key - S3 object key
• version_id - S3 version ID (optional)
• size_bytes - Actual uploaded size (optional)
• content_type - Content type (optional)
• etag - S3 ETag (optional)
• checksum_sha256 - SHA256 checksum (optional)
• corpus_id - Corpus ID to finalize under (optional)

Returns:

FinalizeArtifactResponse

Example:

After uploading to S3

response = await client.v1.artifact.finalize( artifact_id=presign_resp.id, bucket=presign_resp.bucket, key=presign_resp.key, corpus_id="corpus-123" ) print(f"Finalize status: {response.success}")