File Handling

class Files:
    def create(
        self,
        file: BinaryIO,
        *,
        filename: Optional[str] = None,
        content_type: Optional[str] = None,
        **params
    ) -> File:
        """
        Upload a file to Replicate.

        Parameters:
        - file: File object or binary data to upload
        - filename: Name for the uploaded file
        - content_type: MIME type of the file

        Returns:
        File object with metadata and URLs
        """

    def get(self, id: str) -> File:
        """
        Get a file by ID.

        Parameters:
        - id: File ID

        Returns:
        File object with metadata and URLs
        """

    def list(self, **params) -> Page[File]:
        """
        List uploaded files.

        Returns:
        Paginated list of File objects
        """

    def delete(self, id: str) -> None:
        """
        Delete a file.

        Parameters:
        - id: File ID to delete
        """

class File:
    id: str
    """The ID of the file."""

    name: str
    """The name of the file."""

    content_type: str
    """The content type of the file."""

    size: int
    """The size of the file in bytes."""

    etag: str
    """The ETag of the file."""

    checksums: Dict[str, str]
    """The checksums of the file."""

    metadata: Dict[str, Any]
    """The metadata of the file."""

    created_at: str
    """The time the file was created."""

    expires_at: Optional[str]
    """The time the file will expire."""

    urls: Dict[str, str]
    """The URLs of the file (get, delete)."""

class FileOutput:
    def read(self) -> bytes:
        """
        Read the entire file content into memory.

        Returns:
        Complete file content as bytes
        """

    @property
    def url(self) -> str:
        """
        Get the URL of the file.

        Returns:
        Direct URL to access the file
        """

    def __iter__(self) -> Iterator[bytes]:
        """
        Iterate over file content in bytes.

        Returns:
        Iterator yielding file content chunks
        """

    async def aread(self) -> bytes:
        """
        Read the entire file content into memory asynchronously.

        Returns:
        Complete file content as bytes
        """

    async def __aiter__(self) -> AsyncIterator[bytes]:
        """
        Asynchronously iterate over file content.

        Returns:
        AsyncIterator yielding file content chunks
        """

FileEncodingStrategy = Literal["base64", "url"]

import replicate

# Upload a local file
with open("input.jpg", "rb") as f:
    file = replicate.files.create(f, filename="input.jpg")

print(f"File ID: {file.id}")
print(f"File URL: {file.urls['get']}")

# Use uploaded file in model
output = replicate.run(
    "andreasjansson/blip-2",
    input={"image": file.urls['get']}
)

import replicate

# Pass file directly to model
with open("mystery.jpg", "rb") as f:
    output = replicate.run(
        "andreasjansson/blip-2",
        input={"image": f}
    )

print(output)  # "an astronaut riding a horse"

import replicate

# Run model that produces file outputs
outputs = replicate.run(
    "black-forest-labs/flux-schnell",
    input={"prompt": "astronaut riding a rocket"}
)

# Handle single file output
if hasattr(outputs, 'read'):
    with open("output.webp", "wb") as f:
        f.write(outputs.read())

# Handle multiple file outputs
if hasattr(outputs, '__iter__'):
    for index, output in enumerate(outputs):
        with open(f"output_{index}.webp", "wb") as f:
            f.write(output.read())

import replicate

# Get file output from prediction
prediction = replicate.predictions.create(
    model="stability-ai/stable-diffusion-3",
    input={"prompt": "beautiful landscape"}
)
prediction.wait()

# Stream file content efficiently
output = prediction.output
with open("landscape.png", "wb") as f:
    for chunk in output:
        f.write(chunk)

# Or access URL directly
print(f"Direct URL: {output.url}")

import replicate

# Upload and examine file metadata
with open("document.pdf", "rb") as f:
    file = replicate.files.create(f, filename="document.pdf")

print(f"File ID: {file.id}")
print(f"Size: {file.size} bytes") 
print(f"Content Type: {file.content_type}")
print(f"Created: {file.created_at}")
print(f"Checksums: {file.checksums}")
print(f"Expires: {file.expires_at}")

# List all uploaded files
files = replicate.files.list()
for file in files.results:
    print(f"{file.name} ({file.size} bytes)")

import replicate

# Upload temporary file
with open("temp_input.jpg", "rb") as f:
    file = replicate.files.create(f)

# Use file in model
output = replicate.run("some-model", input={"image": file.urls['get']})

# Clean up file
replicate.files.delete(file.id)

import replicate

# Disable FileOutput for legacy compatibility
output = replicate.run(
    "black-forest-labs/flux-schnell",
    input={"prompt": "astronaut"},
    use_file_output=False
)

# Output will be URL strings instead of FileOutput objects
print(type(output))  # <class 'str'>
print(output)        # https://replicate.delivery/...

import replicate

# Model that produces multiple outputs
outputs = replicate.run(
    "pixray/text2image",
    input={"prompts": "sunset over mountains"}
)

# Iterate and save each output
for index, output in enumerate(outputs):
    filename = f"output_{index}.png"
    
    # Method 1: Direct read
    with open(filename, "wb") as f:
        f.write(output.read())
    
    # Method 2: Streaming (more memory efficient)
    with open(f"stream_{index}.png", "wb") as f:
        for chunk in output:
            f.write(chunk)
    
    print(f"Saved {filename} ({output.size} bytes)")

import replicate
import hashlib

# Upload file with validation
with open("input.jpg", "rb") as f:
    content = f.read()
    expected_md5 = hashlib.md5(content).hexdigest()
    
    f.seek(0)  # Reset file pointer
    file = replicate.files.create(f, filename="input.jpg")

# Verify checksum
if 'md5' in file.checksums:
    if file.checksums['md5'] == expected_md5:
        print("File uploaded successfully and verified")
    else:
        print("Checksum mismatch!")