Files

get, list, all, iter, uploads and downloads use V1: only V1 has a file's person, company or opportunity, and it works under any API version pin. get_download_url uses V2 when it can (one request for the details and a signed URL) and V1 otherwise. feed/iter_feed list every file visible to you, newest first, through V2 (Affinity API 2026-09-17+).

Bases: FileSearchMixin

Service for managing files attached to entities.

Source code in affinity/services/v1_only.py

class EntityFileService(FileSearchMixin):
    """Service for managing files attached to entities."""

    def __init__(self, client: HTTPClient):
        self._client = client

    def _validate_exactly_one_target(
        self,
        *,
        person_id: PersonId | None,
        company_id: CompanyId | None,
        opportunity_id: OpportunityId | None,
    ) -> None:
        targets = [person_id, company_id, opportunity_id]
        count = sum(1 for t in targets if t is not None)
        if count == 1:
            return
        if count == 0:
            raise ValueError("Exactly one of person_id, company_id, or opportunity_id is required")
        raise ValueError("Only one of person_id, company_id, or opportunity_id may be provided")

    def list(
        self,
        *,
        person_id: PersonId | None = None,
        company_id: CompanyId | None = None,
        opportunity_id: OpportunityId | None = None,
        page_size: int | None = None,
        page_token: str | None = None,
    ) -> PaginatedResponse[EntityFile]:
        """Get files attached to an entity."""
        self._validate_exactly_one_target(
            person_id=person_id,
            company_id=company_id,
            opportunity_id=opportunity_id,
        )
        params: dict[str, Any] = {}
        if person_id is not None:
            params["person_id"] = int(person_id)
        if company_id is not None:
            params["organization_id"] = int(company_id)
        if opportunity_id is not None:
            params["opportunity_id"] = int(opportunity_id)
        if page_size is not None:
            params["page_size"] = page_size
        if page_token is not None:
            params["page_token"] = page_token

        data = self._client.get("/entity-files", params=params or None, v1=True)
        items = (
            data.get("entity_files")
            or data.get("entityFiles")
            or data.get("files")
            or data.get("data", [])
        )
        if not isinstance(items, list):
            items = []
        return PaginatedResponse[EntityFile](
            data=[EntityFile.model_validate(f) for f in items],
            next_page_token=data.get("next_page_token") or data.get("nextPageToken"),
        )

    def get(self, file_id: FileId) -> EntityFile:
        """A file's details, including the person, company or opportunity it belongs to (V1;
        works under any API version). :meth:`get_download_url` adds a download link."""
        data = self._client.get(f"/entity-files/{file_id}", v1=True)
        return EntityFile.model_validate(data)

    def feed(
        self,
        *,
        created_after: datetime | None = None,
        created_before: datetime | None = None,
        creator_id: UserId | None = None,
        limit: int | None = None,
        total_count: bool = False,
        cursor: str | None = None,
    ) -> PaginatedResponse[FileV2]:
        """One page of the files visible to the caller across the organization, newest first
        (V2, Affinity API 2026-09-17+). V2 can't filter by person, company or opportunity: use
        :meth:`list` for one entity's files. ``limit`` is 1-100."""
        check_cursor_alone(cursor, created_after, created_before, creator_id, limit, total_count)
        if cursor is not None:
            return page_of(
                FileV2, self._client.get_url(cursor, min_api_version=FILES_V2_MIN_API_VERSION)
            )
        params = _file_params(
            created_after=created_after,
            created_before=created_before,
            creator_id=creator_id,
            limit=limit,
            total_count=total_count,
        )
        return page_of(
            FileV2,
            self._client.get(
                "/files", params=params or None, min_api_version=FILES_V2_MIN_API_VERSION
            ),
        )

    def iter_feed(
        self,
        *,
        created_after: datetime | None = None,
        created_before: datetime | None = None,
        creator_id: UserId | None = None,
        limit: int | None = None,
    ) -> Iterator[FileV2]:
        """All files matching the filters (every page); see :meth:`feed`."""

        def fetch_page(next_url: str | None) -> PaginatedResponse[FileV2]:
            if next_url:
                return self.feed(cursor=next_url)
            return self.feed(
                created_after=created_after,
                created_before=created_before,
                creator_id=creator_id,
                limit=limit,
            )

        return PageIterator(fetch_page)

    def download(
        self,
        file_id: FileId,
        *,
        timeout: httpx.Timeout | float | None = None,
        deadline_seconds: float | None = None,
    ) -> bytes:
        """Download file content."""
        return self._client.download_file(
            f"/entity-files/download/{file_id}",
            v1=True,
            timeout=timeout,
            deadline_seconds=deadline_seconds,
        )

    def get_download_url(
        self,
        file_id: FileId,
        *,
        timeout: httpx.Timeout | float | None = None,
    ) -> PresignedUrl:
        """
        Get a presigned download URL for a file without downloading its content.

        The returned URL is valid for approximately 60 seconds and can be
        fetched without authentication (it's self-authenticating via signature).

        Args:
            file_id: The entity file ID
            timeout: Optional request timeout

        Returns:
            PresignedUrl with the URL, file metadata, and expiration info

        Raises:
            AffinityError: If the API doesn't return a redirect URL
        """
        # V2 answers with the details and a signed URL in one request (API 2026-09-17+, beta);
        # a client pinned older, or a V2 refusal, uses V1's details and redirect instead
        try:
            v2 = FileV2.model_validate(
                self._client.get(f"/files/{int(file_id)}", min_api_version=FILES_V2_MIN_API_VERSION)
            )
            if v2.download_url:
                return _presigned(v2.download_url, v2.id, v2.name, v2.size, v2.content_type)
        except (ApiVersionTooOldError, AuthorizationError, NotFoundError):
            pass
        file_meta = self.get(file_id)

        url = self._client.get_redirect_url(
            f"/entity-files/download/{file_id}",
            v1=True,
            timeout=timeout,
        )
        if not url:
            raise AffinityError(
                f"Failed to get presigned URL for file {file_id}: no redirect returned"
            )

        return _presigned(url, file_id, file_meta.name, file_meta.size, file_meta.content_type)

    def download_stream(
        self,
        file_id: FileId,
        *,
        chunk_size: int = 65_536,
        on_progress: ProgressCallback | None = None,
        timeout: httpx.Timeout | float | None = None,
        deadline_seconds: float | None = None,
    ) -> Iterator[bytes]:
        """Stream-download file content in chunks."""
        return self._client.stream_download(
            f"/entity-files/download/{file_id}",
            v1=True,
            chunk_size=chunk_size,
            on_progress=on_progress,
            timeout=timeout,
            deadline_seconds=deadline_seconds,
        )

    def download_stream_with_info(
        self,
        file_id: FileId,
        *,
        chunk_size: int = 65_536,
        on_progress: ProgressCallback | None = None,
        timeout: httpx.Timeout | float | None = None,
        deadline_seconds: float | None = None,
    ) -> DownloadedFile:
        """
        Stream-download a file and return response metadata (headers/filename/size).

        Notes:
        - `filename` is derived from `Content-Disposition` when present.
        - If the server does not provide a filename, callers can fall back to
          `files.get(file_id).name`.
        """
        return self._client.stream_download_with_info(
            f"/entity-files/download/{file_id}",
            v1=True,
            chunk_size=chunk_size,
            on_progress=on_progress,
            timeout=timeout,
            deadline_seconds=deadline_seconds,
        )

    def download_to(
        self,
        file_id: FileId,
        path: str | Path,
        *,
        overwrite: bool = False,
        chunk_size: int = 65_536,
        on_progress: ProgressCallback | None = None,
        timeout: httpx.Timeout | float | None = None,
        deadline_seconds: float | None = None,
    ) -> Path:
        """
        Download a file to disk.

        Args:
            file_id: The entity file id
            path: Destination path
            overwrite: If False, raises FileExistsError when path exists
            chunk_size: Bytes per chunk

        Returns:
            The destination path
        """
        target = Path(path)
        if target.exists() and not overwrite:
            raise FileExistsError(str(target))

        try:
            with target.open("wb") as f:
                for chunk in self.download_stream(
                    file_id,
                    chunk_size=chunk_size,
                    on_progress=on_progress,
                    timeout=timeout,
                    deadline_seconds=deadline_seconds,
                ):
                    f.write(chunk)
        except Exception:
            # Clean up partial file on error
            if target.exists():
                target.unlink()
            raise

        return target

    def _post_upload(
        self,
        files: dict[str, Any],
        *,
        person_id: PersonId | None,
        company_id: CompanyId | None,
        opportunity_id: OpportunityId | None,
    ) -> dict[str, Any]:
        self._validate_exactly_one_target(
            person_id=person_id,
            company_id=company_id,
            opportunity_id=opportunity_id,
        )
        return self._client.upload_file(
            "/entity-files",
            files=files,
            data=_upload_target_data(
                person_id=person_id,
                company_id=company_id,
                opportunity_id=opportunity_id,
            ),
            v1=True,
        )

    def _post_upload_path(
        self,
        path: str | Path,
        *,
        person_id: PersonId | None,
        company_id: CompanyId | None,
        opportunity_id: OpportunityId | None,
        filename: str | None,
        content_type: str | None,
        on_progress: ProgressCallback | None,
    ) -> dict[str, Any]:
        self._validate_exactly_one_target(
            person_id=person_id,
            company_id=company_id,
            opportunity_id=opportunity_id,
        )

        p = Path(path)
        upload_filename = filename or p.name
        guessed, _ = mimetypes.guess_type(upload_filename)
        final_content_type = content_type or guessed or "application/octet-stream"
        total = p.stat().st_size

        if on_progress:
            on_progress(0, total, phase="upload")

        with p.open("rb") as f:
            result = self._post_upload(
                files={"file": (upload_filename, f, final_content_type)},
                person_id=person_id,
                company_id=company_id,
                opportunity_id=opportunity_id,
            )

        if on_progress:
            on_progress(total, total, phase="upload")

        return result

    def _post_upload_bytes(
        self,
        data: bytes,
        filename: str,
        *,
        person_id: PersonId | None,
        company_id: CompanyId | None,
        opportunity_id: OpportunityId | None,
        content_type: str | None,
        on_progress: ProgressCallback | None,
    ) -> dict[str, Any]:
        self._validate_exactly_one_target(
            person_id=person_id,
            company_id=company_id,
            opportunity_id=opportunity_id,
        )

        guessed, _ = mimetypes.guess_type(filename)
        final_content_type = content_type or guessed or "application/octet-stream"
        total = len(data)

        if on_progress:
            on_progress(0, total, phase="upload")

        result = self._post_upload(
            files={"file": (filename, data, final_content_type)},
            person_id=person_id,
            company_id=company_id,
            opportunity_id=opportunity_id,
        )

        if on_progress:
            on_progress(total, total, phase="upload")

        return result

    def upload(
        self,
        files: dict[str, Any],
        *,
        person_id: PersonId | None = None,
        company_id: CompanyId | None = None,
        opportunity_id: OpportunityId | None = None,
    ) -> bool:
        """
        Upload files to an entity.

        Args:
            files: Multipart files mapping, e.g. `{"file": (filename, content, content_type)}`
            person_id: Person to attach to
            company_id: Company to attach to
            opportunity_id: Opportunity to attach to

        Returns:
            True when the upload succeeded (the response's `success` flag when
            present). Use `upload_returning_files()` to get the created file records.
        """
        return _upload_succeeded(
            self._post_upload(
                files,
                person_id=person_id,
                company_id=company_id,
                opportunity_id=opportunity_id,
            )
        )

    def upload_returning_files(
        self,
        files: dict[str, Any],
        *,
        person_id: PersonId | None = None,
        company_id: CompanyId | None = None,
        opportunity_id: OpportunityId | None = None,
    ) -> builtins.list[EntityFile]:
        """
        Upload files to an entity and return the created file records.

        Same request as `upload()`; returns the `entity_files` from the response,
        so the new file ids can be used with `get()`/`download()` straight away.

        Returns:
            The created files. Empty if the API returned no `entity_files` (older
            responses were just `{"success": true}`). Never raises after a successful
            (2xx) upload: an entry that fails validation is returned as a partial
            record (unvalidated; fields that failed are None) when it has an id,
            and skipped otherwise.
        """
        return _parse_uploaded_files(
            self._post_upload(
                files,
                person_id=person_id,
                company_id=company_id,
                opportunity_id=opportunity_id,
            )
        )

    def upload_path(
        self,
        path: str | Path,
        *,
        person_id: PersonId | None = None,
        company_id: CompanyId | None = None,
        opportunity_id: OpportunityId | None = None,
        filename: str | None = None,
        content_type: str | None = None,
        on_progress: ProgressCallback | None = None,
    ) -> bool:
        """
        Upload a file from disk.

        Notes:
        - Returns True when the upload succeeded. Use `upload_path_returning_files()`
          to get the created file record (with its id).
        - Progress reporting is best-effort for uploads (start/end only).
        """
        return _upload_succeeded(
            self._post_upload_path(
                path,
                person_id=person_id,
                company_id=company_id,
                opportunity_id=opportunity_id,
                filename=filename,
                content_type=content_type,
                on_progress=on_progress,
            )
        )

    def upload_path_returning_files(
        self,
        path: str | Path,
        *,
        person_id: PersonId | None = None,
        company_id: CompanyId | None = None,
        opportunity_id: OpportunityId | None = None,
        filename: str | None = None,
        content_type: str | None = None,
        on_progress: ProgressCallback | None = None,
    ) -> builtins.list[EntityFile]:
        """
        Upload a file from disk and return the created file records.

        Same as `upload_path()`, but returns the response's `entity_files`
        (see `upload_returning_files()` for the parsing rules).
        """
        return _parse_uploaded_files(
            self._post_upload_path(
                path,
                person_id=person_id,
                company_id=company_id,
                opportunity_id=opportunity_id,
                filename=filename,
                content_type=content_type,
                on_progress=on_progress,
            )
        )

    def upload_bytes(
        self,
        data: bytes,
        filename: str,
        *,
        person_id: PersonId | None = None,
        company_id: CompanyId | None = None,
        opportunity_id: OpportunityId | None = None,
        content_type: str | None = None,
        on_progress: ProgressCallback | None = None,
    ) -> bool:
        """
        Upload in-memory bytes as a file.

        Notes:
        - Returns True when the upload succeeded. Use `upload_bytes_returning_files()`
          to get the created file record (with its id).
        - Progress reporting is best-effort for uploads (start/end only).
        """
        return _upload_succeeded(
            self._post_upload_bytes(
                data,
                filename,
                person_id=person_id,
                company_id=company_id,
                opportunity_id=opportunity_id,
                content_type=content_type,
                on_progress=on_progress,
            )
        )

    def upload_bytes_returning_files(
        self,
        data: bytes,
        filename: str,
        *,
        person_id: PersonId | None = None,
        company_id: CompanyId | None = None,
        opportunity_id: OpportunityId | None = None,
        content_type: str | None = None,
        on_progress: ProgressCallback | None = None,
    ) -> builtins.list[EntityFile]:
        """
        Upload in-memory bytes as a file and return the created file records.

        Same as `upload_bytes()`, but returns the response's `entity_files`
        (see `upload_returning_files()` for the parsing rules).
        """
        return _parse_uploaded_files(
            self._post_upload_bytes(
                data,
                filename,
                person_id=person_id,
                company_id=company_id,
                opportunity_id=opportunity_id,
                content_type=content_type,
                on_progress=on_progress,
            )
        )

    def all(
        self,
        *,
        person_id: PersonId | None = None,
        company_id: CompanyId | None = None,
        opportunity_id: OpportunityId | None = None,
    ) -> Iterator[EntityFile]:
        """Iterate through all files for an entity with automatic pagination."""
        self._validate_exactly_one_target(
            person_id=person_id,
            company_id=company_id,
            opportunity_id=opportunity_id,
        )

        page_token: str | None = None
        while True:
            page = self.list(
                person_id=person_id,
                company_id=company_id,
                opportunity_id=opportunity_id,
                page_token=page_token,
            )
            yield from page.data
            if not page.has_next:
                break
            page_token = page.next_page_token

    def iter(
        self,
        *,
        person_id: PersonId | None = None,
        company_id: CompanyId | None = None,
        opportunity_id: OpportunityId | None = None,
    ) -> Iterator[EntityFile]:
        """Auto-paginate all files (alias for `all()`)."""
        return self.all(
            person_id=person_id,
            company_id=company_id,
            opportunity_id=opportunity_id,
        )

all(*, person_id: PersonId | None = None, company_id: CompanyId | None = None, opportunity_id: OpportunityId | None = None) -> Iterator[EntityFile]

Iterate through all files for an entity with automatic pagination.

Source code in affinity/services/v1_only.py

def all(
    self,
    *,
    person_id: PersonId | None = None,
    company_id: CompanyId | None = None,
    opportunity_id: OpportunityId | None = None,
) -> Iterator[EntityFile]:
    """Iterate through all files for an entity with automatic pagination."""
    self._validate_exactly_one_target(
        person_id=person_id,
        company_id=company_id,
        opportunity_id=opportunity_id,
    )

    page_token: str | None = None
    while True:
        page = self.list(
            person_id=person_id,
            company_id=company_id,
            opportunity_id=opportunity_id,
            page_token=page_token,
        )
        yield from page.data
        if not page.has_next:
            break
        page_token = page.next_page_token

download(file_id: FileId, *, timeout: httpx.Timeout | float | None = None, deadline_seconds: float | None = None) -> bytes

Download file content.

Source code in affinity/services/v1_only.py

def download(
    self,
    file_id: FileId,
    *,
    timeout: httpx.Timeout | float | None = None,
    deadline_seconds: float | None = None,
) -> bytes:
    """Download file content."""
    return self._client.download_file(
        f"/entity-files/download/{file_id}",
        v1=True,
        timeout=timeout,
        deadline_seconds=deadline_seconds,
    )

download_stream(file_id: FileId, *, chunk_size: int = 65536, on_progress: ProgressCallback | None = None, timeout: httpx.Timeout | float | None = None, deadline_seconds: float | None = None) -> Iterator[bytes]

Stream-download file content in chunks.

Source code in affinity/services/v1_only.py

def download_stream(
    self,
    file_id: FileId,
    *,
    chunk_size: int = 65_536,
    on_progress: ProgressCallback | None = None,
    timeout: httpx.Timeout | float | None = None,
    deadline_seconds: float | None = None,
) -> Iterator[bytes]:
    """Stream-download file content in chunks."""
    return self._client.stream_download(
        f"/entity-files/download/{file_id}",
        v1=True,
        chunk_size=chunk_size,
        on_progress=on_progress,
        timeout=timeout,
        deadline_seconds=deadline_seconds,
    )

download_stream_with_info(file_id: FileId, *, chunk_size: int = 65536, on_progress: ProgressCallback | None = None, timeout: httpx.Timeout | float | None = None, deadline_seconds: float | None = None) -> DownloadedFile

Stream-download a file and return response metadata (headers/filename/size).

Notes: - filename is derived from Content-Disposition when present. - If the server does not provide a filename, callers can fall back to files.get(file_id).name.

Source code in affinity/services/v1_only.py

def download_stream_with_info(
    self,
    file_id: FileId,
    *,
    chunk_size: int = 65_536,
    on_progress: ProgressCallback | None = None,
    timeout: httpx.Timeout | float | None = None,
    deadline_seconds: float | None = None,
) -> DownloadedFile:
    """
    Stream-download a file and return response metadata (headers/filename/size).

    Notes:
    - `filename` is derived from `Content-Disposition` when present.
    - If the server does not provide a filename, callers can fall back to
      `files.get(file_id).name`.
    """
    return self._client.stream_download_with_info(
        f"/entity-files/download/{file_id}",
        v1=True,
        chunk_size=chunk_size,
        on_progress=on_progress,
        timeout=timeout,
        deadline_seconds=deadline_seconds,
    )

download_to(file_id: FileId, path: str | Path, *, overwrite: bool = False, chunk_size: int = 65536, on_progress: ProgressCallback | None = None, timeout: httpx.Timeout | float | None = None, deadline_seconds: float | None = None) -> Path

Download a file to disk.

Parameters:

Name	Type	Description	Default
file_id	FileId	The entity file id	required
path	str | Path	Destination path	required
overwrite	bool	If False, raises FileExistsError when path exists	False
chunk_size	int	Bytes per chunk	65536

Returns:

Type	Description
Path	The destination path

Source code in affinity/services/v1_only.py

def download_to(
    self,
    file_id: FileId,
    path: str | Path,
    *,
    overwrite: bool = False,
    chunk_size: int = 65_536,
    on_progress: ProgressCallback | None = None,
    timeout: httpx.Timeout | float | None = None,
    deadline_seconds: float | None = None,
) -> Path:
    """
    Download a file to disk.

    Args:
        file_id: The entity file id
        path: Destination path
        overwrite: If False, raises FileExistsError when path exists
        chunk_size: Bytes per chunk

    Returns:
        The destination path
    """
    target = Path(path)
    if target.exists() and not overwrite:
        raise FileExistsError(str(target))

    try:
        with target.open("wb") as f:
            for chunk in self.download_stream(
                file_id,
                chunk_size=chunk_size,
                on_progress=on_progress,
                timeout=timeout,
                deadline_seconds=deadline_seconds,
            ):
                f.write(chunk)
    except Exception:
        # Clean up partial file on error
        if target.exists():
            target.unlink()
        raise

    return target

feed(*, created_after: datetime | None = None, created_before: datetime | None = None, creator_id: UserId | None = None, limit: int | None = None, total_count: bool = False, cursor: str | None = None) -> PaginatedResponse[FileV2]

One page of the files visible to the caller across the organization, newest first (V2, Affinity API 2026-09-17+). V2 can't filter by person, company or opportunity: use :meth:list for one entity's files. limit is 1-100.

Source code in affinity/services/v1_only.py

def feed(
    self,
    *,
    created_after: datetime | None = None,
    created_before: datetime | None = None,
    creator_id: UserId | None = None,
    limit: int | None = None,
    total_count: bool = False,
    cursor: str | None = None,
) -> PaginatedResponse[FileV2]:
    """One page of the files visible to the caller across the organization, newest first
    (V2, Affinity API 2026-09-17+). V2 can't filter by person, company or opportunity: use
    :meth:`list` for one entity's files. ``limit`` is 1-100."""
    check_cursor_alone(cursor, created_after, created_before, creator_id, limit, total_count)
    if cursor is not None:
        return page_of(
            FileV2, self._client.get_url(cursor, min_api_version=FILES_V2_MIN_API_VERSION)
        )
    params = _file_params(
        created_after=created_after,
        created_before=created_before,
        creator_id=creator_id,
        limit=limit,
        total_count=total_count,
    )
    return page_of(
        FileV2,
        self._client.get(
            "/files", params=params or None, min_api_version=FILES_V2_MIN_API_VERSION
        ),
    )

get(file_id: FileId) -> EntityFile

A file's details, including the person, company or opportunity it belongs to (V1; works under any API version). :meth:get_download_url adds a download link.

Source code in affinity/services/v1_only.py

def get(self, file_id: FileId) -> EntityFile:
    """A file's details, including the person, company or opportunity it belongs to (V1;
    works under any API version). :meth:`get_download_url` adds a download link."""
    data = self._client.get(f"/entity-files/{file_id}", v1=True)
    return EntityFile.model_validate(data)

get_download_url(file_id: FileId, *, timeout: httpx.Timeout | float | None = None) -> PresignedUrl

Get a presigned download URL for a file without downloading its content.

The returned URL is valid for approximately 60 seconds and can be fetched without authentication (it's self-authenticating via signature).

Parameters:

Name	Type	Description	Default
file_id	FileId	The entity file ID	required
timeout	Timeout | float | None	Optional request timeout	None

Returns:

Type	Description
PresignedUrl	PresignedUrl with the URL, file metadata, and expiration info

Raises:

Type	Description
AffinityError	If the API doesn't return a redirect URL

Source code in affinity/services/v1_only.py

def get_download_url(
    self,
    file_id: FileId,
    *,
    timeout: httpx.Timeout | float | None = None,
) -> PresignedUrl:
    """
    Get a presigned download URL for a file without downloading its content.

    The returned URL is valid for approximately 60 seconds and can be
    fetched without authentication (it's self-authenticating via signature).

    Args:
        file_id: The entity file ID
        timeout: Optional request timeout

    Returns:
        PresignedUrl with the URL, file metadata, and expiration info

    Raises:
        AffinityError: If the API doesn't return a redirect URL
    """
    # V2 answers with the details and a signed URL in one request (API 2026-09-17+, beta);
    # a client pinned older, or a V2 refusal, uses V1's details and redirect instead
    try:
        v2 = FileV2.model_validate(
            self._client.get(f"/files/{int(file_id)}", min_api_version=FILES_V2_MIN_API_VERSION)
        )
        if v2.download_url:
            return _presigned(v2.download_url, v2.id, v2.name, v2.size, v2.content_type)
    except (ApiVersionTooOldError, AuthorizationError, NotFoundError):
        pass
    file_meta = self.get(file_id)

    url = self._client.get_redirect_url(
        f"/entity-files/download/{file_id}",
        v1=True,
        timeout=timeout,
    )
    if not url:
        raise AffinityError(
            f"Failed to get presigned URL for file {file_id}: no redirect returned"
        )

    return _presigned(url, file_id, file_meta.name, file_meta.size, file_meta.content_type)

iter(*, person_id: PersonId | None = None, company_id: CompanyId | None = None, opportunity_id: OpportunityId | None = None) -> Iterator[EntityFile]

Auto-paginate all files (alias for all()).

Source code in affinity/services/v1_only.py

def iter(
    self,
    *,
    person_id: PersonId | None = None,
    company_id: CompanyId | None = None,
    opportunity_id: OpportunityId | None = None,
) -> Iterator[EntityFile]:
    """Auto-paginate all files (alias for `all()`)."""
    return self.all(
        person_id=person_id,
        company_id=company_id,
        opportunity_id=opportunity_id,
    )

iter_feed(*, created_after: datetime | None = None, created_before: datetime | None = None, creator_id: UserId | None = None, limit: int | None = None) -> Iterator[FileV2]

All files matching the filters (every page); see :meth:feed.

Source code in affinity/services/v1_only.py

def iter_feed(
    self,
    *,
    created_after: datetime | None = None,
    created_before: datetime | None = None,
    creator_id: UserId | None = None,
    limit: int | None = None,
) -> Iterator[FileV2]:
    """All files matching the filters (every page); see :meth:`feed`."""

    def fetch_page(next_url: str | None) -> PaginatedResponse[FileV2]:
        if next_url:
            return self.feed(cursor=next_url)
        return self.feed(
            created_after=created_after,
            created_before=created_before,
            creator_id=creator_id,
            limit=limit,
        )

    return PageIterator(fetch_page)

list(*, person_id: PersonId | None = None, company_id: CompanyId | None = None, opportunity_id: OpportunityId | None = None, page_size: int | None = None, page_token: str | None = None) -> PaginatedResponse[EntityFile]

Get files attached to an entity.

Source code in affinity/services/v1_only.py

def list(
    self,
    *,
    person_id: PersonId | None = None,
    company_id: CompanyId | None = None,
    opportunity_id: OpportunityId | None = None,
    page_size: int | None = None,
    page_token: str | None = None,
) -> PaginatedResponse[EntityFile]:
    """Get files attached to an entity."""
    self._validate_exactly_one_target(
        person_id=person_id,
        company_id=company_id,
        opportunity_id=opportunity_id,
    )
    params: dict[str, Any] = {}
    if person_id is not None:
        params["person_id"] = int(person_id)
    if company_id is not None:
        params["organization_id"] = int(company_id)
    if opportunity_id is not None:
        params["opportunity_id"] = int(opportunity_id)
    if page_size is not None:
        params["page_size"] = page_size
    if page_token is not None:
        params["page_token"] = page_token

    data = self._client.get("/entity-files", params=params or None, v1=True)
    items = (
        data.get("entity_files")
        or data.get("entityFiles")
        or data.get("files")
        or data.get("data", [])
    )
    if not isinstance(items, list):
        items = []
    return PaginatedResponse[EntityFile](
        data=[EntityFile.model_validate(f) for f in items],
        next_page_token=data.get("next_page_token") or data.get("nextPageToken"),
    )

upload(files: dict[str, Any], *, person_id: PersonId | None = None, company_id: CompanyId | None = None, opportunity_id: OpportunityId | None = None) -> bool

Upload files to an entity.

Parameters:

Name	Type	Description	Default
files	dict[str, Any]	Multipart files mapping, e.g. {"file": (filename, content, content_type)}	required
person_id	PersonId | None	Person to attach to	None
company_id	CompanyId | None	Company to attach to	None
opportunity_id	OpportunityId | None	Opportunity to attach to	None

Returns:

Type	Description
bool	True when the upload succeeded (the response's success flag when
bool	present). Use upload_returning_files() to get the created file records.

Source code in affinity/services/v1_only.py

def upload(
    self,
    files: dict[str, Any],
    *,
    person_id: PersonId | None = None,
    company_id: CompanyId | None = None,
    opportunity_id: OpportunityId | None = None,
) -> bool:
    """
    Upload files to an entity.

    Args:
        files: Multipart files mapping, e.g. `{"file": (filename, content, content_type)}`
        person_id: Person to attach to
        company_id: Company to attach to
        opportunity_id: Opportunity to attach to

    Returns:
        True when the upload succeeded (the response's `success` flag when
        present). Use `upload_returning_files()` to get the created file records.
    """
    return _upload_succeeded(
        self._post_upload(
            files,
            person_id=person_id,
            company_id=company_id,
            opportunity_id=opportunity_id,
        )
    )

upload_bytes(data: bytes, filename: str, *, person_id: PersonId | None = None, company_id: CompanyId | None = None, opportunity_id: OpportunityId | None = None, content_type: str | None = None, on_progress: ProgressCallback | None = None) -> bool

Upload in-memory bytes as a file.

Notes: - Returns True when the upload succeeded. Use upload_bytes_returning_files() to get the created file record (with its id). - Progress reporting is best-effort for uploads (start/end only).

Source code in affinity/services/v1_only.py

def upload_bytes(
    self,
    data: bytes,
    filename: str,
    *,
    person_id: PersonId | None = None,
    company_id: CompanyId | None = None,
    opportunity_id: OpportunityId | None = None,
    content_type: str | None = None,
    on_progress: ProgressCallback | None = None,
) -> bool:
    """
    Upload in-memory bytes as a file.

    Notes:
    - Returns True when the upload succeeded. Use `upload_bytes_returning_files()`
      to get the created file record (with its id).
    - Progress reporting is best-effort for uploads (start/end only).
    """
    return _upload_succeeded(
        self._post_upload_bytes(
            data,
            filename,
            person_id=person_id,
            company_id=company_id,
            opportunity_id=opportunity_id,
            content_type=content_type,
            on_progress=on_progress,
        )
    )

upload_bytes_returning_files(data: bytes, filename: str, *, person_id: PersonId | None = None, company_id: CompanyId | None = None, opportunity_id: OpportunityId | None = None, content_type: str | None = None, on_progress: ProgressCallback | None = None) -> builtins.list[EntityFile]

Upload in-memory bytes as a file and return the created file records.

Same as upload_bytes(), but returns the response's entity_files (see upload_returning_files() for the parsing rules).

Source code in affinity/services/v1_only.py

def upload_bytes_returning_files(
    self,
    data: bytes,
    filename: str,
    *,
    person_id: PersonId | None = None,
    company_id: CompanyId | None = None,
    opportunity_id: OpportunityId | None = None,
    content_type: str | None = None,
    on_progress: ProgressCallback | None = None,
) -> builtins.list[EntityFile]:
    """
    Upload in-memory bytes as a file and return the created file records.

    Same as `upload_bytes()`, but returns the response's `entity_files`
    (see `upload_returning_files()` for the parsing rules).
    """
    return _parse_uploaded_files(
        self._post_upload_bytes(
            data,
            filename,
            person_id=person_id,
            company_id=company_id,
            opportunity_id=opportunity_id,
            content_type=content_type,
            on_progress=on_progress,
        )
    )

upload_path(path: str | Path, *, person_id: PersonId | None = None, company_id: CompanyId | None = None, opportunity_id: OpportunityId | None = None, filename: str | None = None, content_type: str | None = None, on_progress: ProgressCallback | None = None) -> bool

Upload a file from disk.

Notes: - Returns True when the upload succeeded. Use upload_path_returning_files() to get the created file record (with its id). - Progress reporting is best-effort for uploads (start/end only).

Source code in affinity/services/v1_only.py

def upload_path(
    self,
    path: str | Path,
    *,
    person_id: PersonId | None = None,
    company_id: CompanyId | None = None,
    opportunity_id: OpportunityId | None = None,
    filename: str | None = None,
    content_type: str | None = None,
    on_progress: ProgressCallback | None = None,
) -> bool:
    """
    Upload a file from disk.

    Notes:
    - Returns True when the upload succeeded. Use `upload_path_returning_files()`
      to get the created file record (with its id).
    - Progress reporting is best-effort for uploads (start/end only).
    """
    return _upload_succeeded(
        self._post_upload_path(
            path,
            person_id=person_id,
            company_id=company_id,
            opportunity_id=opportunity_id,
            filename=filename,
            content_type=content_type,
            on_progress=on_progress,
        )
    )

upload_path_returning_files(path: str | Path, *, person_id: PersonId | None = None, company_id: CompanyId | None = None, opportunity_id: OpportunityId | None = None, filename: str | None = None, content_type: str | None = None, on_progress: ProgressCallback | None = None) -> builtins.list[EntityFile]

Upload a file from disk and return the created file records.

Same as upload_path(), but returns the response's entity_files (see upload_returning_files() for the parsing rules).

Source code in affinity/services/v1_only.py

def upload_path_returning_files(
    self,
    path: str | Path,
    *,
    person_id: PersonId | None = None,
    company_id: CompanyId | None = None,
    opportunity_id: OpportunityId | None = None,
    filename: str | None = None,
    content_type: str | None = None,
    on_progress: ProgressCallback | None = None,
) -> builtins.list[EntityFile]:
    """
    Upload a file from disk and return the created file records.

    Same as `upload_path()`, but returns the response's `entity_files`
    (see `upload_returning_files()` for the parsing rules).
    """
    return _parse_uploaded_files(
        self._post_upload_path(
            path,
            person_id=person_id,
            company_id=company_id,
            opportunity_id=opportunity_id,
            filename=filename,
            content_type=content_type,
            on_progress=on_progress,
        )
    )

upload_returning_files(files: dict[str, Any], *, person_id: PersonId | None = None, company_id: CompanyId | None = None, opportunity_id: OpportunityId | None = None) -> builtins.list[EntityFile]

Upload files to an entity and return the created file records.

Same request as upload(); returns the entity_files from the response, so the new file ids can be used with get()/download() straight away.

Returns:

Type	Description
list[EntityFile]	The created files. Empty if the API returned no entity_files (older
list[EntityFile]	responses were just {"success": true}). Never raises after a successful
list[EntityFile]	(2xx) upload: an entry that fails validation is returned as a partial
list[EntityFile]	record (unvalidated; fields that failed are None) when it has an id,
list[EntityFile]	and skipped otherwise.

Source code in affinity/services/v1_only.py

def upload_returning_files(
    self,
    files: dict[str, Any],
    *,
    person_id: PersonId | None = None,
    company_id: CompanyId | None = None,
    opportunity_id: OpportunityId | None = None,
) -> builtins.list[EntityFile]:
    """
    Upload files to an entity and return the created file records.

    Same request as `upload()`; returns the `entity_files` from the response,
    so the new file ids can be used with `get()`/`download()` straight away.

    Returns:
        The created files. Empty if the API returned no `entity_files` (older
        responses were just `{"success": true}`). Never raises after a successful
        (2xx) upload: an entry that fails validation is returned as a partial
        record (unvalidated; fields that failed are None) when it has an id,
        and skipped otherwise.
    """
    return _parse_uploaded_files(
        self._post_upload(
            files,
            person_id=person_id,
            company_id=company_id,
            opportunity_id=opportunity_id,
        )
    )