Files (V1)

Service for managing files attached to entities.

Source code in affinity/services/v1_only.py

class EntityFileService:
    """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,
    ) -> V1PaginatedResponse[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:
            params["person_id"] = int(person_id)
        if company_id:
            params["organization_id"] = int(company_id)
        if opportunity_id:
            params["opportunity_id"] = int(opportunity_id)
        if page_size:
            params["page_size"] = page_size
        if page_token:
            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 V1PaginatedResponse[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:
        """Get file metadata."""
        data = self._client.get(f"/entity-files/{file_id}", v1=True)
        return EntityFile.model_validate(data)

    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 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))

        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)

        return target

    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: Dict of filename to file-like object
            person_id: Person to attach to
            company_id: Company to attach to
            opportunity_id: Opportunity to attach to

        Returns:
            List of created file records
        """
        self._validate_exactly_one_target(
            person_id=person_id,
            company_id=company_id,
            opportunity_id=opportunity_id,
        )
        data: dict[str, Any] = {}
        if person_id:
            data["person_id"] = int(person_id)
        if company_id:
            data["organization_id"] = int(company_id)
        if opportunity_id:
            data["opportunity_id"] = int(opportunity_id)

        result = self._client.upload_file(
            "/entity-files",
            files=files,
            data=data,
            v1=True,
        )
        if "success" in result:
            return bool(result.get("success"))
        # If the API returns something else on success (e.g., created object),
        # treat any 2xx JSON response as success (4xx/5xx raise earlier).
        return True

    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 only a boolean because the API returns `{"success": true}` for uploads.
        - Progress reporting is best-effort for uploads (start/end only).
        """
        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:
            ok = self.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 ok

    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 only a boolean because the API returns `{"success": true}` for uploads.
        - Progress reporting is best-effort for uploads (start/end only).
        """
        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")

        ok = self.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 ok

    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))

    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)

    return target

get(file_id: FileId) -> EntityFile

Get file metadata.

Source code in affinity/services/v1_only.py

def get(self, file_id: FileId) -> EntityFile:
    """Get file metadata."""
    data = self._client.get(f"/entity-files/{file_id}", v1=True)
    return EntityFile.model_validate(data)

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,
    )

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) -> V1PaginatedResponse[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,
) -> V1PaginatedResponse[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:
        params["person_id"] = int(person_id)
    if company_id:
        params["organization_id"] = int(company_id)
    if opportunity_id:
        params["opportunity_id"] = int(opportunity_id)
    if page_size:
        params["page_size"] = page_size
    if page_token:
        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 V1PaginatedResponse[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]	Dict of filename to file-like object	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	List of 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: Dict of filename to file-like object
        person_id: Person to attach to
        company_id: Company to attach to
        opportunity_id: Opportunity to attach to

    Returns:
        List of created file records
    """
    self._validate_exactly_one_target(
        person_id=person_id,
        company_id=company_id,
        opportunity_id=opportunity_id,
    )
    data: dict[str, Any] = {}
    if person_id:
        data["person_id"] = int(person_id)
    if company_id:
        data["organization_id"] = int(company_id)
    if opportunity_id:
        data["opportunity_id"] = int(opportunity_id)

    result = self._client.upload_file(
        "/entity-files",
        files=files,
        data=data,
        v1=True,
    )
    if "success" in result:
        return bool(result.get("success"))
    # If the API returns something else on success (e.g., created object),
    # treat any 2xx JSON response as success (4xx/5xx raise earlier).
    return True

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 only a boolean because the API returns {"success": true} for uploads. - 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 only a boolean because the API returns `{"success": true}` for uploads.
    - Progress reporting is best-effort for uploads (start/end only).
    """
    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")

    ok = self.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 ok

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 only a boolean because the API returns {"success": true} for uploads. - 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 only a boolean because the API returns `{"success": true}` for uploads.
    - Progress reporting is best-effort for uploads (start/end only).
    """
    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:
        ok = self.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 ok