Client

Affinity

Synchronous Affinity API client.

Provides access to all Affinity API functionality with a clean, Pythonic interface. Uses V2 API where available, falls back to V1 for operations not yet supported in V2.

Example

from affinity import Affinity

# Initialize with API key
client = Affinity(api_key="your-api-key")

# Use as context manager for automatic cleanup
with Affinity(api_key="your-api-key") as client:
    # Get all companies
    for company in client.companies.all():
        print(company.name)

    # Get a specific person with field data
    person = client.persons.get(
        PersonId(12345),
        field_types=["enriched", "global"]
    )

    # Add a company to a list
    entries = client.lists.entries(ListId(789))
    entry = entries.add_company(CompanyId(456))

    # Update field values
    entries.update_field_value(
        entry.id,
        FieldId(101),
        "New value"
    )

Attributes:

Name	Type	Description
companies	CompanyService	Company (organization) operations, incl. semantic_search
persons	PersonService	Person (contact) operations
opportunities	OpportunityService	Opportunity operations
lists	ListService	List and list entry operations
tasks	TaskService	Long-running task (e.g. merge) status
notes	NoteService	Note operations, incl. keyword search
reminders	ReminderService	Reminder operations
webhooks	WebhookService	Webhook subscription operations
interactions	InteractionService	Interaction (email, meeting, etc.) operations
transcripts	TranscriptService	Meeting transcripts (AI Notetaker)
teams	TeamService	Team operations
users	UserService	User operations
feedback	FeedbackService	Send feedback to Affinity
fields	FieldService	Custom field operations
field_values	FieldValueService	Field value operations (V1 service)
field_value_changes	FieldValueChangesService	Field value change history operations
files	EntityFileService	Entity file operations, incl. keyword search of file contents
relationships	RelationshipStrengthService	Relationship strength queries
auth	AuthService	Authentication info (whoami)
rate_limits	RateLimitService	Rate limit status

Source code in affinity/client.py

class Affinity:
    """
    Synchronous Affinity API client.

    Provides access to all Affinity API functionality with a clean,
    Pythonic interface. Uses V2 API where available, falls back to V1
    for operations not yet supported in V2.

    Example:
        ```python
        from affinity import Affinity

        # Initialize with API key
        client = Affinity(api_key="your-api-key")

        # Use as context manager for automatic cleanup
        with Affinity(api_key="your-api-key") as client:
            # Get all companies
            for company in client.companies.all():
                print(company.name)

            # Get a specific person with field data
            person = client.persons.get(
                PersonId(12345),
                field_types=["enriched", "global"]
            )

            # Add a company to a list
            entries = client.lists.entries(ListId(789))
            entry = entries.add_company(CompanyId(456))

            # Update field values
            entries.update_field_value(
                entry.id,
                FieldId(101),
                "New value"
            )
        ```

    Attributes:
        companies: Company (organization) operations, incl. ``semantic_search``
        persons: Person (contact) operations
        opportunities: Opportunity operations
        lists: List and list entry operations
        tasks: Long-running task (e.g. merge) status
        notes: Note operations, incl. keyword ``search``
        reminders: Reminder operations
        webhooks: Webhook subscription operations
        interactions: Interaction (email, meeting, etc.) operations
        transcripts: Meeting transcripts (AI Notetaker)
        teams: Team operations
        users: User operations
        feedback: Send feedback to Affinity
        fields: Custom field operations
        field_values: Field value operations (V1 service)
        field_value_changes: Field value change history operations
        files: Entity file operations, incl. keyword ``search`` of file contents
        relationships: Relationship strength queries
        auth: Authentication info (``whoami``)
        rate_limits: Rate limit status
    """

    def __init__(
        self,
        api_key: str,
        *,
        v1_base_url: str = V1_BASE_URL,
        v2_base_url: str = V2_BASE_URL,
        v1_auth_mode: Literal["bearer", "basic"] = "bearer",
        transport: httpx.BaseTransport | None = None,
        async_transport: httpx.AsyncBaseTransport | None = None,
        enable_beta_endpoints: bool = False,
        allow_insecure_download_redirects: bool = False,
        expected_v2_version: str | None = None,
        affinity_api_version: str | None = None,
        timeout: float = 30.0,
        max_retries: int = 3,
        enable_cache: bool = False,
        cache_ttl: float = 300.0,
        log_requests: bool = False,
        on_request: RequestHook | None = None,
        on_response: ResponseHook | None = None,
        on_error: ErrorHook | None = None,
        on_event: AnyEventHook | None = None,
        hook_error_policy: Literal["swallow", "raise"] = "swallow",
        policies: Policies | None = None,
    ):
        """
        Initialize the Affinity client.

        Args:
            api_key: Your Affinity API key
            v1_base_url: V1 API base URL (default: https://api.affinity.co)
            v2_base_url: V2 API base URL (default: https://api.affinity.co/v2)
            v1_auth_mode: Auth mode for V1 API ("bearer" or "basic")
            transport: Optional `httpx` transport (advanced; useful for mocking in tests)
            async_transport: Optional async `httpx` transport (advanced; useful for mocking in
                tests)
            enable_beta_endpoints: Enable beta V2 endpoints
            allow_insecure_download_redirects: Allow `http://` redirects for file downloads.
                Not recommended; prefer HTTPS-only downloads.
            expected_v2_version: Deprecated. Expected V2 API version, used only to label
                diagnostics (a mismatch with the echoed version is logged once). Use
                ``affinity_api_version`` to pin the version instead.
            affinity_api_version: Affinity V2 API version to request, sent as the
                ``X-Affinity-Api-Version`` header on every V2 request (e.g. "2026-09-17",
                or "current" for the newest). None (default), "auto" or "key-default"
                send no header, so the API key's default version applies. V1 requests
                never carry it. Invalid values raise ``ConfigurationError``; a version
                Affinity rejects raises ``UnsupportedApiVersionError``.
            timeout: Request timeout in seconds
            max_retries: Maximum retries for rate-limited requests
            enable_cache: Enable response caching for field metadata
            cache_ttl: Cache TTL in seconds
            log_requests: Log all HTTP requests (for debugging)
            on_request: Hook called before each request (DX-008)
            on_response: Hook called after each response (DX-008)
            on_error: Hook called when a request raises (DX-008)
            on_event: Event hook called for request/response lifecycle events (DX-008)
            hook_error_policy: What to do if hooks raise ("swallow" or "raise")
            policies: Client policies (e.g., disable writes)

        Raises:
            ConfigurationError: A hook is an ``async def`` function (use ``AsyncAffinity``).
        """
        _reject_async_hooks(
            on_request=on_request, on_response=on_response, on_error=on_error, on_event=on_event
        )
        config = ClientConfig(
            api_key=api_key,
            v1_base_url=v1_base_url,
            v2_base_url=v2_base_url,
            v1_auth_mode=v1_auth_mode,
            transport=transport,
            async_transport=async_transport,
            enable_beta_endpoints=enable_beta_endpoints,
            allow_insecure_download_redirects=allow_insecure_download_redirects,
            expected_v2_version=expected_v2_version,
            affinity_api_version=affinity_api_version,
            timeout=timeout,
            max_retries=max_retries,
            enable_cache=enable_cache,
            cache_ttl=cache_ttl,
            log_requests=log_requests,
            on_request=on_request,
            on_response=on_response,
            on_error=on_error,
            on_event=on_event,
            hook_error_policy=hook_error_policy,
            policies=policies or Policies(),
        )
        self._http = HTTPClient(config)

        # Resource management tracking
        self._closed = False
        self._entered_context = False

        # Initialize services
        self._companies: CompanyService | None = None
        self._persons: PersonService | None = None
        self._lists: ListService | None = None
        self._opportunities: OpportunityService | None = None
        self._tasks: TaskService | None = None
        self._notes: NoteService | None = None
        self._transcripts: TranscriptService | None = None
        self._reminders: ReminderService | None = None
        self._webhooks: WebhookService | None = None
        self._teams: TeamService | None = None
        self._users: UserService | None = None
        self._feedback: FeedbackService | None = None
        self._interactions: InteractionService | None = None
        self._fields: FieldService | None = None
        self._field_values: FieldValueService | None = None
        self._field_value_changes: FieldValueChangesService | None = None
        self._files: EntityFileService | None = None
        self._relationships: RelationshipStrengthService | None = None
        self._auth: AuthService | None = None
        self._rate_limits: RateLimitService | None = None

    @classmethod
    def from_env(
        cls,
        *,
        env_var: str = _DEFAULT_API_KEY_ENV_VAR,
        load_dotenv: bool = False,
        dotenv_path: str | os.PathLike[str] | None = None,
        dotenv_override: bool = False,
        transport: httpx.BaseTransport | None = None,
        async_transport: httpx.AsyncBaseTransport | None = None,
        policies: Policies | None = None,
        on_event: AnyEventHook | None = None,
        hook_error_policy: Literal["swallow", "raise"] = "swallow",
        **kwargs: Any,
    ) -> Affinity:
        """
        Create a client using an API key from the environment.

        By default, reads `AFFINITY_API_KEY`. For local development, you can optionally
        load a `.env` file (requires `python-dotenv`).
        """
        api_key = _api_key_from_env(
            env_var=env_var,
            load_dotenv=load_dotenv,
            dotenv_path=dotenv_path,
            dotenv_override=dotenv_override,
        )
        return cls(
            api_key=api_key,
            transport=transport,
            async_transport=async_transport,
            policies=policies,
            on_event=on_event,
            hook_error_policy=hook_error_policy,
            **kwargs,
        )

    @classmethod
    def read_only(
        cls,
        api_key: str,
        **kwargs: Any,
    ) -> Affinity:
        """
        Create a read-only client that blocks all write operations.

        This is a convenience factory equivalent to:
            Affinity(api_key, policies=Policies(write=WritePolicy.DENY))

        For env-based key resolution, use read_only_from_env() instead.

        Args:
            api_key: Affinity API key (required).
            **kwargs: Additional arguments passed to Affinity constructor
                (e.g., v1_base_url, v2_base_url, timeout).

        Returns:
            Affinity client configured to reject write operations.

        Example:
            >>> with Affinity.read_only("your-key") as client:
            ...     companies = client.companies.list()
            ...     client.companies.create(...)  # Raises WriteNotAllowedError
        """
        if "policies" in kwargs:
            raise ValueError(
                "Cannot specify 'policies' with read_only(); "
                "use Affinity() constructor directly for custom policies"
            )
        return cls(
            api_key=api_key,
            policies=Policies(write=WritePolicy.DENY),
            **kwargs,
        )

    @classmethod
    def read_only_from_env(
        cls,
        *,
        env_var: str = _DEFAULT_API_KEY_ENV_VAR,
        **kwargs: Any,
    ) -> Affinity:
        """
        Create a read-only client using API key from environment.

        Equivalent to Affinity.from_env() with WritePolicy.DENY.

        Args:
            env_var: Environment variable name (default: AFFINITY_API_KEY).
            **kwargs: Additional arguments passed to from_env().

        Example:
            >>> with Affinity.read_only_from_env() as client:
            ...     companies = client.companies.list()
        """
        if "policies" in kwargs:
            raise ValueError(
                "Cannot specify 'policies' with read_only_from_env(); "
                "use Affinity.from_env() directly for custom policies"
            )
        return cls.from_env(
            env_var=env_var,
            policies=Policies(write=WritePolicy.DENY),
            **kwargs,
        )

    def __enter__(self) -> Affinity:
        self._entered_context = True
        return self

    def __exit__(self, *args: Any) -> None:
        self.close()

    def close(self) -> None:
        """Close the HTTP client and release resources."""
        if not self._closed:
            self._http.close()
            self._closed = True

    def __del__(self) -> None:
        """Warn if client was not properly closed."""
        # Use getattr to handle case where __init__ failed before setting _closed
        if not getattr(self, "_closed", True) and not getattr(self, "_entered_context", True):
            warnings.warn(
                "Affinity client was not closed. "
                "Use 'with Affinity.from_env() as client:' "
                "or call client.close() when done.",
                ResourceWarning,
                stacklevel=2,
            )
            # Still close to prevent actual resource leaks
            with contextlib.suppress(Exception):
                self.close()

    # =========================================================================
    # Service Properties (lazy initialization)
    # =========================================================================

    @property
    def companies(self) -> CompanyService:
        """Company (organization) operations."""
        if self._companies is None:
            self._companies = CompanyService(self._http)
        return self._companies

    @property
    def persons(self) -> PersonService:
        """Person (contact) operations."""
        if self._persons is None:
            self._persons = PersonService(self._http)
        return self._persons

    @property
    def lists(self) -> ListService:
        """List operations."""
        if self._lists is None:
            self._lists = ListService(self._http)
        return self._lists

    @property
    def opportunities(self) -> OpportunityService:
        """Opportunity operations."""
        if self._opportunities is None:
            self._opportunities = OpportunityService(self._http)
        return self._opportunities

    @property
    def tasks(self) -> TaskService:
        """Long-running task operations (polling, waiting)."""
        if self._tasks is None:
            self._tasks = TaskService(self._http)
        return self._tasks

    @property
    def transcripts(self) -> TranscriptService:
        """Meeting transcripts (AI Notetaker, V2)."""
        if self._transcripts is None:
            self._transcripts = TranscriptService(self._http)
        return self._transcripts

    @property
    def notes(self) -> NoteService:
        """Note operations."""
        if self._notes is None:
            self._notes = NoteService(self._http)
        return self._notes

    @property
    def reminders(self) -> ReminderService:
        """Reminder operations."""
        if self._reminders is None:
            self._reminders = ReminderService(self._http)
        return self._reminders

    @property
    def webhooks(self) -> WebhookService:
        """Webhook subscription operations."""
        if self._webhooks is None:
            self._webhooks = WebhookService(self._http)
        return self._webhooks

    @property
    def teams(self) -> TeamService:
        """Teams (V2, beta; Affinity API 2026-09-17+)."""
        if self._teams is None:
            self._teams = TeamService(self._http)
        return self._teams

    @property
    def users(self) -> UserService:
        """The organization's internal users (V2, beta)."""
        if self._users is None:
            self._users = UserService(self._http)
        return self._users

    @property
    def feedback(self) -> FeedbackService:
        """Feedback to Affinity's product team."""
        if self._feedback is None:
            self._feedback = FeedbackService(self._http)
        return self._feedback

    @property
    def interactions(self) -> InteractionService:
        """Interaction operations."""
        if self._interactions is None:
            self._interactions = InteractionService(self._http)
        return self._interactions

    @property
    def fields(self) -> FieldService:
        """Custom field operations."""
        if self._fields is None:
            self._fields = FieldService(self._http)
        return self._fields

    @property
    def field_values(self) -> FieldValueService:
        """Field value operations."""
        if self._field_values is None:
            self._field_values = FieldValueService(self._http)
        return self._field_values

    @property
    def field_value_changes(self) -> FieldValueChangesService:
        """Field value change history queries."""
        if self._field_value_changes is None:
            self._field_value_changes = FieldValueChangesService(self._http)
        return self._field_value_changes

    @property
    def files(self) -> EntityFileService:
        """Entity file operations."""
        if self._files is None:
            self._files = EntityFileService(self._http)
        return self._files

    @property
    def relationships(self) -> RelationshipStrengthService:
        """Relationship strength queries."""
        if self._relationships is None:
            self._relationships = RelationshipStrengthService(self._http)
        return self._relationships

    @property
    def auth(self) -> AuthService:
        """Authentication info."""
        if self._auth is None:
            self._auth = AuthService(self._http)
        return self._auth

    @property
    def rate_limits(self) -> RateLimitService:
        """Unified rate limit information (version-agnostic)."""
        if self._rate_limits is None:
            self._rate_limits = RateLimitService(self._http)
        return self._rate_limits

    # =========================================================================
    # Utility Methods
    # =========================================================================

    def clear_cache(self) -> None:
        """Clear the response cache."""
        if self._http.cache:
            self._http.cache.clear()

    def whoami(self) -> WhoAmI:
        """Convenience wrapper for `client.auth.whoami()`."""
        return self.auth.whoami()

    @property
    def affinity_api_version(self) -> str | None:
        """Configured V2 API version (``X-Affinity-Api-Version``); None = the key's default."""
        return self._http.affinity_api_version

    @property
    def affinity_api_versions_seen(self) -> frozenset[str]:
        """V2 API versions Affinity reported answering with so far (incl. cache hits)."""
        return self._http.affinity_api_versions_seen

    def require_api_version(self, minimum: str, *, operation: str) -> None:
        """
        Raise ``ApiVersionTooOldError`` if this client is pinned to a version older than
        ``minimum`` (nothing is sent). Unpinned clients always pass: version-gated calls send
        their minimum version themselves.
        """
        self._http.require_api_version(minimum, operation=operation)

    @property
    def affinity_api_versions_per_operation(self) -> frozenset[str]:
        """Versions that answered calls needing a newer API version than the key's default.

        Such calls (e.g. company/person field writes) send their minimum version when the
        client isn't pinned. A subset of ``affinity_api_versions_seen``.
        """
        return self._http.affinity_api_versions_per_operation

    def get_key_default_api_version(self) -> str | None:
        """The API key's default V2 API version (one unversioned V2 request)."""
        return self._http.probe_key_default_api_version()

affinity_api_version: str | None property

Configured V2 API version (X-Affinity-Api-Version); None = the key's default.

affinity_api_versions_per_operation: frozenset[str] property

Versions that answered calls needing a newer API version than the key's default.

Such calls (e.g. company/person field writes) send their minimum version when the client isn't pinned. A subset of affinity_api_versions_seen.

affinity_api_versions_seen: frozenset[str] property

V2 API versions Affinity reported answering with so far (incl. cache hits).

auth: AuthService property

Authentication info.

companies: CompanyService property

Company (organization) operations.

feedback: FeedbackService property

Feedback to Affinity's product team.

field_value_changes: FieldValueChangesService property

Field value change history queries.

field_values: FieldValueService property

Field value operations.

fields: FieldService property

Custom field operations.

files: EntityFileService property

Entity file operations.

interactions: InteractionService property

Interaction operations.

lists: ListService property

List operations.

notes: NoteService property

Note operations.

opportunities: OpportunityService property

Opportunity operations.

persons: PersonService property

Person (contact) operations.

rate_limits: RateLimitService property

Unified rate limit information (version-agnostic).

relationships: RelationshipStrengthService property

Relationship strength queries.

reminders: ReminderService property

Reminder operations.

tasks: TaskService property

Long-running task operations (polling, waiting).

teams: TeamService property

Teams (V2, beta; Affinity API 2026-09-17+).

transcripts: TranscriptService property

Meeting transcripts (AI Notetaker, V2).

users: UserService property

The organization's internal users (V2, beta).

webhooks: WebhookService property

Webhook subscription operations.

__del__() -> None

Warn if client was not properly closed.

Source code in affinity/client.py

def __del__(self) -> None:
    """Warn if client was not properly closed."""
    # Use getattr to handle case where __init__ failed before setting _closed
    if not getattr(self, "_closed", True) and not getattr(self, "_entered_context", True):
        warnings.warn(
            "Affinity client was not closed. "
            "Use 'with Affinity.from_env() as client:' "
            "or call client.close() when done.",
            ResourceWarning,
            stacklevel=2,
        )
        # Still close to prevent actual resource leaks
        with contextlib.suppress(Exception):
            self.close()

__init__(api_key: str, *, v1_base_url: str = V1_BASE_URL, v2_base_url: str = V2_BASE_URL, v1_auth_mode: Literal['bearer', 'basic'] = 'bearer', transport: httpx.BaseTransport | None = None, async_transport: httpx.AsyncBaseTransport | None = None, enable_beta_endpoints: bool = False, allow_insecure_download_redirects: bool = False, expected_v2_version: str | None = None, affinity_api_version: str | None = None, timeout: float = 30.0, max_retries: int = 3, enable_cache: bool = False, cache_ttl: float = 300.0, log_requests: bool = False, on_request: RequestHook | None = None, on_response: ResponseHook | None = None, on_error: ErrorHook | None = None, on_event: AnyEventHook | None = None, hook_error_policy: Literal['swallow', 'raise'] = 'swallow', policies: Policies | None = None)

Initialize the Affinity client.

Parameters:

Name	Type	Description	Default
api_key	str	Your Affinity API key	required
v1_base_url	str	V1 API base URL (default: https://api.affinity.co)	V1_BASE_URL
v2_base_url	str	V2 API base URL (default: https://api.affinity.co/v2)	V2_BASE_URL
v1_auth_mode	Literal['bearer', 'basic']	Auth mode for V1 API ("bearer" or "basic")	'bearer'
transport	BaseTransport | None	Optional httpx transport (advanced; useful for mocking in tests)	None
async_transport	AsyncBaseTransport | None	Optional async httpx transport (advanced; useful for mocking in tests)	None
enable_beta_endpoints	bool	Enable beta V2 endpoints	False
allow_insecure_download_redirects	bool	Allow http:// redirects for file downloads. Not recommended; prefer HTTPS-only downloads.	False
expected_v2_version	str | None	Deprecated. Expected V2 API version, used only to label diagnostics (a mismatch with the echoed version is logged once). Use affinity_api_version to pin the version instead.	None
affinity_api_version	str | None	Affinity V2 API version to request, sent as the X-Affinity-Api-Version header on every V2 request (e.g. "2026-09-17", or "current" for the newest). None (default), "auto" or "key-default" send no header, so the API key's default version applies. V1 requests never carry it. Invalid values raise ConfigurationError; a version Affinity rejects raises UnsupportedApiVersionError.	None
timeout	float	Request timeout in seconds	30.0
max_retries	int	Maximum retries for rate-limited requests	3
enable_cache	bool	Enable response caching for field metadata	False
cache_ttl	float	Cache TTL in seconds	300.0
log_requests	bool	Log all HTTP requests (for debugging)	False
on_request	RequestHook | None	Hook called before each request (DX-008)	None
on_response	ResponseHook | None	Hook called after each response (DX-008)	None
on_error	ErrorHook | None	Hook called when a request raises (DX-008)	None
on_event	AnyEventHook | None	Event hook called for request/response lifecycle events (DX-008)	None
hook_error_policy	Literal['swallow', 'raise']	What to do if hooks raise ("swallow" or "raise")	'swallow'
policies	Policies | None	Client policies (e.g., disable writes)	None

Raises:

Type	Description
ConfigurationError	A hook is an async def function (use AsyncAffinity).

Source code in affinity/client.py

def __init__(
    self,
    api_key: str,
    *,
    v1_base_url: str = V1_BASE_URL,
    v2_base_url: str = V2_BASE_URL,
    v1_auth_mode: Literal["bearer", "basic"] = "bearer",
    transport: httpx.BaseTransport | None = None,
    async_transport: httpx.AsyncBaseTransport | None = None,
    enable_beta_endpoints: bool = False,
    allow_insecure_download_redirects: bool = False,
    expected_v2_version: str | None = None,
    affinity_api_version: str | None = None,
    timeout: float = 30.0,
    max_retries: int = 3,
    enable_cache: bool = False,
    cache_ttl: float = 300.0,
    log_requests: bool = False,
    on_request: RequestHook | None = None,
    on_response: ResponseHook | None = None,
    on_error: ErrorHook | None = None,
    on_event: AnyEventHook | None = None,
    hook_error_policy: Literal["swallow", "raise"] = "swallow",
    policies: Policies | None = None,
):
    """
    Initialize the Affinity client.

    Args:
        api_key: Your Affinity API key
        v1_base_url: V1 API base URL (default: https://api.affinity.co)
        v2_base_url: V2 API base URL (default: https://api.affinity.co/v2)
        v1_auth_mode: Auth mode for V1 API ("bearer" or "basic")
        transport: Optional `httpx` transport (advanced; useful for mocking in tests)
        async_transport: Optional async `httpx` transport (advanced; useful for mocking in
            tests)
        enable_beta_endpoints: Enable beta V2 endpoints
        allow_insecure_download_redirects: Allow `http://` redirects for file downloads.
            Not recommended; prefer HTTPS-only downloads.
        expected_v2_version: Deprecated. Expected V2 API version, used only to label
            diagnostics (a mismatch with the echoed version is logged once). Use
            ``affinity_api_version`` to pin the version instead.
        affinity_api_version: Affinity V2 API version to request, sent as the
            ``X-Affinity-Api-Version`` header on every V2 request (e.g. "2026-09-17",
            or "current" for the newest). None (default), "auto" or "key-default"
            send no header, so the API key's default version applies. V1 requests
            never carry it. Invalid values raise ``ConfigurationError``; a version
            Affinity rejects raises ``UnsupportedApiVersionError``.
        timeout: Request timeout in seconds
        max_retries: Maximum retries for rate-limited requests
        enable_cache: Enable response caching for field metadata
        cache_ttl: Cache TTL in seconds
        log_requests: Log all HTTP requests (for debugging)
        on_request: Hook called before each request (DX-008)
        on_response: Hook called after each response (DX-008)
        on_error: Hook called when a request raises (DX-008)
        on_event: Event hook called for request/response lifecycle events (DX-008)
        hook_error_policy: What to do if hooks raise ("swallow" or "raise")
        policies: Client policies (e.g., disable writes)

    Raises:
        ConfigurationError: A hook is an ``async def`` function (use ``AsyncAffinity``).
    """
    _reject_async_hooks(
        on_request=on_request, on_response=on_response, on_error=on_error, on_event=on_event
    )
    config = ClientConfig(
        api_key=api_key,
        v1_base_url=v1_base_url,
        v2_base_url=v2_base_url,
        v1_auth_mode=v1_auth_mode,
        transport=transport,
        async_transport=async_transport,
        enable_beta_endpoints=enable_beta_endpoints,
        allow_insecure_download_redirects=allow_insecure_download_redirects,
        expected_v2_version=expected_v2_version,
        affinity_api_version=affinity_api_version,
        timeout=timeout,
        max_retries=max_retries,
        enable_cache=enable_cache,
        cache_ttl=cache_ttl,
        log_requests=log_requests,
        on_request=on_request,
        on_response=on_response,
        on_error=on_error,
        on_event=on_event,
        hook_error_policy=hook_error_policy,
        policies=policies or Policies(),
    )
    self._http = HTTPClient(config)

    # Resource management tracking
    self._closed = False
    self._entered_context = False

    # Initialize services
    self._companies: CompanyService | None = None
    self._persons: PersonService | None = None
    self._lists: ListService | None = None
    self._opportunities: OpportunityService | None = None
    self._tasks: TaskService | None = None
    self._notes: NoteService | None = None
    self._transcripts: TranscriptService | None = None
    self._reminders: ReminderService | None = None
    self._webhooks: WebhookService | None = None
    self._teams: TeamService | None = None
    self._users: UserService | None = None
    self._feedback: FeedbackService | None = None
    self._interactions: InteractionService | None = None
    self._fields: FieldService | None = None
    self._field_values: FieldValueService | None = None
    self._field_value_changes: FieldValueChangesService | None = None
    self._files: EntityFileService | None = None
    self._relationships: RelationshipStrengthService | None = None
    self._auth: AuthService | None = None
    self._rate_limits: RateLimitService | None = None

clear_cache() -> None

Clear the response cache.

Source code in affinity/client.py

def clear_cache(self) -> None:
    """Clear the response cache."""
    if self._http.cache:
        self._http.cache.clear()

close() -> None

Close the HTTP client and release resources.

Source code in affinity/client.py

def close(self) -> None:
    """Close the HTTP client and release resources."""
    if not self._closed:
        self._http.close()
        self._closed = True

from_env(*, env_var: str = _DEFAULT_API_KEY_ENV_VAR, load_dotenv: bool = False, dotenv_path: str | os.PathLike[str] | None = None, dotenv_override: bool = False, transport: httpx.BaseTransport | None = None, async_transport: httpx.AsyncBaseTransport | None = None, policies: Policies | None = None, on_event: AnyEventHook | None = None, hook_error_policy: Literal['swallow', 'raise'] = 'swallow', **kwargs: Any) -> Affinity classmethod

Create a client using an API key from the environment.

By default, reads AFFINITY_API_KEY. For local development, you can optionally load a .env file (requires python-dotenv).

Source code in affinity/client.py

@classmethod
def from_env(
    cls,
    *,
    env_var: str = _DEFAULT_API_KEY_ENV_VAR,
    load_dotenv: bool = False,
    dotenv_path: str | os.PathLike[str] | None = None,
    dotenv_override: bool = False,
    transport: httpx.BaseTransport | None = None,
    async_transport: httpx.AsyncBaseTransport | None = None,
    policies: Policies | None = None,
    on_event: AnyEventHook | None = None,
    hook_error_policy: Literal["swallow", "raise"] = "swallow",
    **kwargs: Any,
) -> Affinity:
    """
    Create a client using an API key from the environment.

    By default, reads `AFFINITY_API_KEY`. For local development, you can optionally
    load a `.env` file (requires `python-dotenv`).
    """
    api_key = _api_key_from_env(
        env_var=env_var,
        load_dotenv=load_dotenv,
        dotenv_path=dotenv_path,
        dotenv_override=dotenv_override,
    )
    return cls(
        api_key=api_key,
        transport=transport,
        async_transport=async_transport,
        policies=policies,
        on_event=on_event,
        hook_error_policy=hook_error_policy,
        **kwargs,
    )

get_key_default_api_version() -> str | None

The API key's default V2 API version (one unversioned V2 request).

Source code in affinity/client.py

def get_key_default_api_version(self) -> str | None:
    """The API key's default V2 API version (one unversioned V2 request)."""
    return self._http.probe_key_default_api_version()

read_only(api_key: str, **kwargs: Any) -> Affinity classmethod

Create a read-only client that blocks all write operations.

This is a convenience factory equivalent to

Affinity(api_key, policies=Policies(write=WritePolicy.DENY))

For env-based key resolution, use read_only_from_env() instead.

Parameters:

Name	Type	Description	Default
api_key	str	Affinity API key (required).	required
**kwargs	Any	Additional arguments passed to Affinity constructor (e.g., v1_base_url, v2_base_url, timeout).	{}

Returns:

Type	Description
Affinity	Affinity client configured to reject write operations.

Example

with Affinity.read_only("your-key") as client: ... companies = client.companies.list() ... client.companies.create(...) # Raises WriteNotAllowedError

Source code in affinity/client.py

@classmethod
def read_only(
    cls,
    api_key: str,
    **kwargs: Any,
) -> Affinity:
    """
    Create a read-only client that blocks all write operations.

    This is a convenience factory equivalent to:
        Affinity(api_key, policies=Policies(write=WritePolicy.DENY))

    For env-based key resolution, use read_only_from_env() instead.

    Args:
        api_key: Affinity API key (required).
        **kwargs: Additional arguments passed to Affinity constructor
            (e.g., v1_base_url, v2_base_url, timeout).

    Returns:
        Affinity client configured to reject write operations.

    Example:
        >>> with Affinity.read_only("your-key") as client:
        ...     companies = client.companies.list()
        ...     client.companies.create(...)  # Raises WriteNotAllowedError
    """
    if "policies" in kwargs:
        raise ValueError(
            "Cannot specify 'policies' with read_only(); "
            "use Affinity() constructor directly for custom policies"
        )
    return cls(
        api_key=api_key,
        policies=Policies(write=WritePolicy.DENY),
        **kwargs,
    )

read_only_from_env(*, env_var: str = _DEFAULT_API_KEY_ENV_VAR, **kwargs: Any) -> Affinity classmethod

Create a read-only client using API key from environment.

Equivalent to Affinity.from_env() with WritePolicy.DENY.

Parameters:

Name	Type	Description	Default
env_var	str	Environment variable name (default: AFFINITY_API_KEY).	_DEFAULT_API_KEY_ENV_VAR
**kwargs	Any	Additional arguments passed to from_env().	{}

Example

with Affinity.read_only_from_env() as client: ... companies = client.companies.list()

Source code in affinity/client.py

@classmethod
def read_only_from_env(
    cls,
    *,
    env_var: str = _DEFAULT_API_KEY_ENV_VAR,
    **kwargs: Any,
) -> Affinity:
    """
    Create a read-only client using API key from environment.

    Equivalent to Affinity.from_env() with WritePolicy.DENY.

    Args:
        env_var: Environment variable name (default: AFFINITY_API_KEY).
        **kwargs: Additional arguments passed to from_env().

    Example:
        >>> with Affinity.read_only_from_env() as client:
        ...     companies = client.companies.list()
    """
    if "policies" in kwargs:
        raise ValueError(
            "Cannot specify 'policies' with read_only_from_env(); "
            "use Affinity.from_env() directly for custom policies"
        )
    return cls.from_env(
        env_var=env_var,
        policies=Policies(write=WritePolicy.DENY),
        **kwargs,
    )

require_api_version(minimum: str, *, operation: str) -> None

Raise ApiVersionTooOldError if this client is pinned to a version older than minimum (nothing is sent). Unpinned clients always pass: version-gated calls send their minimum version themselves.

Source code in affinity/client.py

def require_api_version(self, minimum: str, *, operation: str) -> None:
    """
    Raise ``ApiVersionTooOldError`` if this client is pinned to a version older than
    ``minimum`` (nothing is sent). Unpinned clients always pass: version-gated calls send
    their minimum version themselves.
    """
    self._http.require_api_version(minimum, operation=operation)

whoami() -> WhoAmI

Convenience wrapper for client.auth.whoami().

Source code in affinity/client.py

def whoami(self) -> WhoAmI:
    """Convenience wrapper for `client.auth.whoami()`."""
    return self.auth.whoami()

AsyncAffinity

Asynchronous Affinity API client.

Same interface as Affinity but with async/await support: the same service attributes (companies, persons, opportunities, lists, tasks, notes, reminders, webhooks, interactions, transcripts, teams, users, feedback, fields, field_values, field_value_changes, files, relationships, auth, rate_limits).

Example

async with AsyncAffinity(api_key="your-key") as client:
    async for company in client.companies.all():
        print(company.name)

Source code in affinity/client.py

class AsyncAffinity:
    """
    Asynchronous Affinity API client.

    Same interface as Affinity but with async/await support: the same service attributes
    (``companies``, ``persons``, ``opportunities``, ``lists``, ``tasks``, ``notes``,
    ``reminders``, ``webhooks``, ``interactions``, ``transcripts``, ``teams``, ``users``,
    ``feedback``, ``fields``, ``field_values``, ``field_value_changes``, ``files``,
    ``relationships``, ``auth``, ``rate_limits``).

    Example:
        ```python
        async with AsyncAffinity(api_key="your-key") as client:
            async for company in client.companies.all():
                print(company.name)
        ```
    """

    def __init__(
        self,
        api_key: str,
        *,
        v1_base_url: str = V1_BASE_URL,
        v2_base_url: str = V2_BASE_URL,
        v1_auth_mode: Literal["bearer", "basic"] = "bearer",
        transport: httpx.BaseTransport | None = None,
        async_transport: httpx.AsyncBaseTransport | None = None,
        enable_beta_endpoints: bool = False,
        allow_insecure_download_redirects: bool = False,
        expected_v2_version: str | None = None,
        affinity_api_version: str | None = None,
        timeout: float = 30.0,
        max_retries: int = 3,
        enable_cache: bool = False,
        cache_ttl: float = 300.0,
        log_requests: bool = False,
        on_request: RequestHook | None = None,
        on_response: ResponseHook | None = None,
        on_error: ErrorHook | None = None,
        on_event: AnyEventHook | None = None,
        hook_error_policy: Literal["swallow", "raise"] = "swallow",
        policies: Policies | None = None,
    ):
        """
        Initialize the async Affinity client.

        Args:
            api_key: Your Affinity API key
            v1_base_url: V1 API base URL (default: https://api.affinity.co)
            v2_base_url: V2 API base URL (default: https://api.affinity.co/v2)
            v1_auth_mode: Auth mode for V1 API ("bearer" or "basic")
            transport: Optional `httpx` transport (advanced; useful for mocking in tests)
            async_transport: Optional async `httpx` transport (advanced; useful for mocking in
                tests)
            enable_beta_endpoints: Enable beta V2 endpoints
            allow_insecure_download_redirects: Allow `http://` redirects for file downloads.
                Not recommended; prefer HTTPS-only downloads.
            expected_v2_version: Deprecated. Expected V2 API version, used only to label
                diagnostics (a mismatch with the echoed version is logged once). Use
                ``affinity_api_version`` to pin the version instead.
            affinity_api_version: Affinity V2 API version to request, sent as the
                ``X-Affinity-Api-Version`` header on every V2 request (e.g. "2026-09-17",
                or "current" for the newest). None (default), "auto" or "key-default"
                send no header, so the API key's default version applies. V1 requests
                never carry it. Invalid values raise ``ConfigurationError``; a version
                Affinity rejects raises ``UnsupportedApiVersionError``.
            timeout: Request timeout in seconds
            max_retries: Maximum retries for rate-limited requests
            enable_cache: Enable response caching for field metadata
            cache_ttl: Cache TTL in seconds
            log_requests: Log all HTTP requests (for debugging)
            on_request: Hook called before each request (DX-008)
            on_response: Hook called after each response (DX-008)
            on_error: Hook called when a request raises (DX-008)
            on_event: Event hook called for request/response lifecycle events (DX-008)
            hook_error_policy: What to do if hooks raise ("swallow" or "raise")
        """
        config = ClientConfig(
            api_key=api_key,
            v1_base_url=v1_base_url,
            v2_base_url=v2_base_url,
            v1_auth_mode=v1_auth_mode,
            transport=transport,
            async_transport=async_transport,
            enable_beta_endpoints=enable_beta_endpoints,
            allow_insecure_download_redirects=allow_insecure_download_redirects,
            expected_v2_version=expected_v2_version,
            affinity_api_version=affinity_api_version,
            timeout=timeout,
            max_retries=max_retries,
            enable_cache=enable_cache,
            cache_ttl=cache_ttl,
            log_requests=log_requests,
            on_request=on_request,
            on_response=on_response,
            on_error=on_error,
            on_event=on_event,
            hook_error_policy=hook_error_policy,
            policies=policies or Policies(),
        )
        self._http = AsyncHTTPClient(config)

        # Resource management tracking
        self._closed = False
        self._entered_context = False

        self._companies: AsyncCompanyService | None = None
        self._persons: AsyncPersonService | None = None
        self._opportunities: AsyncOpportunityService | None = None
        self._lists: AsyncListService | None = None
        self._tasks: AsyncTaskService | None = None
        self._notes: AsyncNoteService | None = None
        self._transcripts: AsyncTranscriptService | None = None
        self._reminders: AsyncReminderService | None = None
        self._webhooks: AsyncWebhookService | None = None
        self._teams: AsyncTeamService | None = None
        self._users: AsyncUserService | None = None
        self._feedback: AsyncFeedbackService | None = None
        self._interactions: AsyncInteractionService | None = None
        self._fields: AsyncFieldService | None = None
        self._field_values: AsyncFieldValueService | None = None
        self._field_value_changes: AsyncFieldValueChangesService | None = None
        self._files: AsyncEntityFileService | None = None
        self._relationships: AsyncRelationshipStrengthService | None = None
        self._auth: AsyncAuthService | None = None
        self._rate_limits: AsyncRateLimitService | None = None

    @classmethod
    def from_env(
        cls,
        *,
        env_var: str = _DEFAULT_API_KEY_ENV_VAR,
        load_dotenv: bool = False,
        dotenv_path: str | os.PathLike[str] | None = None,
        dotenv_override: bool = False,
        transport: httpx.BaseTransport | None = None,
        async_transport: httpx.AsyncBaseTransport | None = None,
        policies: Policies | None = None,
        on_event: AnyEventHook | None = None,
        hook_error_policy: Literal["swallow", "raise"] = "swallow",
        **kwargs: Any,
    ) -> AsyncAffinity:
        """
        Create an async client using an API key from the environment.

        By default, reads `AFFINITY_API_KEY`. For local development, you can optionally
        load a `.env` file (requires `python-dotenv`).
        """
        api_key = _api_key_from_env(
            env_var=env_var,
            load_dotenv=load_dotenv,
            dotenv_path=dotenv_path,
            dotenv_override=dotenv_override,
        )
        return cls(
            api_key=api_key,
            transport=transport,
            async_transport=async_transport,
            policies=policies,
            on_event=on_event,
            hook_error_policy=hook_error_policy,
            **kwargs,
        )

    @classmethod
    def read_only(
        cls,
        api_key: str,
        **kwargs: Any,
    ) -> AsyncAffinity:
        """
        Create a read-only async client that blocks all write operations.

        See Affinity.read_only() for details.
        """
        if "policies" in kwargs:
            raise ValueError(
                "Cannot specify 'policies' with read_only(); "
                "use AsyncAffinity() constructor directly for custom policies"
            )
        return cls(
            api_key=api_key,
            policies=Policies(write=WritePolicy.DENY),
            **kwargs,
        )

    @classmethod
    def read_only_from_env(
        cls,
        *,
        env_var: str = _DEFAULT_API_KEY_ENV_VAR,
        **kwargs: Any,
    ) -> AsyncAffinity:
        """
        Create a read-only async client using API key from environment.

        See Affinity.read_only_from_env() for details.
        """
        if "policies" in kwargs:
            raise ValueError(
                "Cannot specify 'policies' with read_only_from_env(); "
                "use AsyncAffinity.from_env() directly for custom policies"
            )
        return cls.from_env(
            env_var=env_var,
            policies=Policies(write=WritePolicy.DENY),
            **kwargs,
        )

    @property
    def companies(self) -> AsyncCompanyService:
        """Company (organization) operations."""
        if self._companies is None:
            self._companies = AsyncCompanyService(self._http)
        return self._companies

    @property
    def persons(self) -> AsyncPersonService:
        """Person (contact) operations."""
        if self._persons is None:
            self._persons = AsyncPersonService(self._http)
        return self._persons

    @property
    def opportunities(self) -> AsyncOpportunityService:
        """Opportunity operations."""
        if self._opportunities is None:
            self._opportunities = AsyncOpportunityService(self._http)
        return self._opportunities

    @property
    def lists(self) -> AsyncListService:
        """List and list entry operations."""
        if self._lists is None:
            self._lists = AsyncListService(self._http)
        return self._lists

    @property
    def tasks(self) -> AsyncTaskService:
        """Long-running task operations (polling, waiting)."""
        if self._tasks is None:
            self._tasks = AsyncTaskService(self._http)
        return self._tasks

    @property
    def transcripts(self) -> AsyncTranscriptService:
        """Meeting transcripts (AI Notetaker, V2)."""
        if self._transcripts is None:
            self._transcripts = AsyncTranscriptService(self._http)
        return self._transcripts

    @property
    def notes(self) -> AsyncNoteService:
        """Note operations."""
        if self._notes is None:
            self._notes = AsyncNoteService(self._http)
        return self._notes

    @property
    def reminders(self) -> AsyncReminderService:
        """Reminder operations."""
        if self._reminders is None:
            self._reminders = AsyncReminderService(self._http)
        return self._reminders

    @property
    def webhooks(self) -> AsyncWebhookService:
        """Webhook subscription operations."""
        if self._webhooks is None:
            self._webhooks = AsyncWebhookService(self._http)
        return self._webhooks

    @property
    def teams(self) -> AsyncTeamService:
        """Teams (V2, beta; Affinity API 2026-09-17+)."""
        if self._teams is None:
            self._teams = AsyncTeamService(self._http)
        return self._teams

    @property
    def users(self) -> AsyncUserService:
        """The organization's internal users (V2, beta)."""
        if self._users is None:
            self._users = AsyncUserService(self._http)
        return self._users

    @property
    def feedback(self) -> AsyncFeedbackService:
        """Feedback to Affinity's product team."""
        if self._feedback is None:
            self._feedback = AsyncFeedbackService(self._http)
        return self._feedback

    @property
    def interactions(self) -> AsyncInteractionService:
        """Interaction operations."""
        if self._interactions is None:
            self._interactions = AsyncInteractionService(self._http)
        return self._interactions

    @property
    def fields(self) -> AsyncFieldService:
        """Custom field operations."""
        if self._fields is None:
            self._fields = AsyncFieldService(self._http)
        return self._fields

    @property
    def field_values(self) -> AsyncFieldValueService:
        """Field value operations."""
        if self._field_values is None:
            self._field_values = AsyncFieldValueService(self._http)
        return self._field_values

    @property
    def field_value_changes(self) -> AsyncFieldValueChangesService:
        """Field value change history queries."""
        if self._field_value_changes is None:
            self._field_value_changes = AsyncFieldValueChangesService(self._http)
        return self._field_value_changes

    @property
    def files(self) -> AsyncEntityFileService:
        """Entity file operations."""
        if self._files is None:
            self._files = AsyncEntityFileService(self._http)
        return self._files

    @property
    def relationships(self) -> AsyncRelationshipStrengthService:
        """Relationship strength queries."""
        if self._relationships is None:
            self._relationships = AsyncRelationshipStrengthService(self._http)
        return self._relationships

    @property
    def auth(self) -> AsyncAuthService:
        """Authentication info."""
        if self._auth is None:
            self._auth = AsyncAuthService(self._http)
        return self._auth

    @property
    def rate_limits(self) -> AsyncRateLimitService:
        """Unified rate limit information (version-agnostic)."""
        if self._rate_limits is None:
            self._rate_limits = AsyncRateLimitService(self._http)
        return self._rate_limits

    async def __aenter__(self) -> AsyncAffinity:
        """Enter an async context and return this client."""
        self._entered_context = True
        return self

    async def __aexit__(self, *args: Any) -> None:
        """Exit the async context and close the underlying HTTP client."""
        await self.close()

    async def close(self) -> None:
        """Close the HTTP client."""
        if not self._closed:
            await self._http.close()
            self._closed = True

    def __del__(self) -> None:
        """Warn if client was not properly closed.

        NOTE: We intentionally do NOT attempt async cleanup here.
        asyncio.create_task() in __del__ is unsafe - the task may be garbage
        collected before completion, leading to resource leaks or errors.
        Users must use context managers or call close() explicitly.
        """
        # Use getattr to handle case where __init__ failed before setting _closed
        if not getattr(self, "_closed", True) and not getattr(self, "_entered_context", True):
            warnings.warn(
                "AsyncAffinity client was not closed. "
                "Use 'async with AsyncAffinity.from_env() as client:' "
                "or call await client.close() when done.",
                ResourceWarning,
                stacklevel=2,
            )

    def clear_cache(self) -> None:
        """Clear the response cache."""
        if self._http.cache:
            self._http.cache.clear()

    async def whoami(self) -> WhoAmI:
        """Convenience wrapper for `client.auth.whoami()`."""
        return await self.auth.whoami()

    @property
    def affinity_api_version(self) -> str | None:
        """Configured V2 API version (``X-Affinity-Api-Version``); None = the key's default."""
        return self._http.affinity_api_version

    @property
    def affinity_api_versions_seen(self) -> frozenset[str]:
        """V2 API versions Affinity reported answering with so far (incl. cache hits)."""
        return self._http.affinity_api_versions_seen

    def require_api_version(self, minimum: str, *, operation: str) -> None:
        """
        Raise ``ApiVersionTooOldError`` if this client is pinned to a version older than
        ``minimum`` (nothing is sent). Unpinned clients always pass: version-gated calls send
        their minimum version themselves.
        """
        self._http.require_api_version(minimum, operation=operation)

    @property
    def affinity_api_versions_per_operation(self) -> frozenset[str]:
        """Versions that answered calls needing a newer API version than the key's default.

        Such calls (e.g. company/person field writes) send their minimum version when the
        client isn't pinned. A subset of ``affinity_api_versions_seen``.
        """
        return self._http.affinity_api_versions_per_operation

    async def get_key_default_api_version(self) -> str | None:
        """The API key's default V2 API version (one unversioned V2 request)."""
        return await self._http.probe_key_default_api_version()

affinity_api_version: str | None property

Configured V2 API version (X-Affinity-Api-Version); None = the key's default.

affinity_api_versions_per_operation: frozenset[str] property

Versions that answered calls needing a newer API version than the key's default.

Such calls (e.g. company/person field writes) send their minimum version when the client isn't pinned. A subset of affinity_api_versions_seen.

affinity_api_versions_seen: frozenset[str] property

V2 API versions Affinity reported answering with so far (incl. cache hits).

auth: AsyncAuthService property

Authentication info.

companies: AsyncCompanyService property

Company (organization) operations.

feedback: AsyncFeedbackService property

Feedback to Affinity's product team.

field_value_changes: AsyncFieldValueChangesService property

Field value change history queries.

field_values: AsyncFieldValueService property

Field value operations.

fields: AsyncFieldService property

Custom field operations.

files: AsyncEntityFileService property

Entity file operations.

interactions: AsyncInteractionService property

Interaction operations.

lists: AsyncListService property

List and list entry operations.

notes: AsyncNoteService property

Note operations.

opportunities: AsyncOpportunityService property

Opportunity operations.

persons: AsyncPersonService property

Person (contact) operations.

rate_limits: AsyncRateLimitService property

Unified rate limit information (version-agnostic).

relationships: AsyncRelationshipStrengthService property

Relationship strength queries.

reminders: AsyncReminderService property

Reminder operations.

tasks: AsyncTaskService property

Long-running task operations (polling, waiting).

teams: AsyncTeamService property

Teams (V2, beta; Affinity API 2026-09-17+).

transcripts: AsyncTranscriptService property

Meeting transcripts (AI Notetaker, V2).

users: AsyncUserService property

The organization's internal users (V2, beta).

webhooks: AsyncWebhookService property

Webhook subscription operations.

__aenter__() -> AsyncAffinity async

Enter an async context and return this client.

Source code in affinity/client.py

async def __aenter__(self) -> AsyncAffinity:
    """Enter an async context and return this client."""
    self._entered_context = True
    return self

__aexit__(*args: Any) -> None async

Exit the async context and close the underlying HTTP client.

Source code in affinity/client.py

async def __aexit__(self, *args: Any) -> None:
    """Exit the async context and close the underlying HTTP client."""
    await self.close()

__del__() -> None

Warn if client was not properly closed.

NOTE: We intentionally do NOT attempt async cleanup here. asyncio.create_task() in del is unsafe - the task may be garbage collected before completion, leading to resource leaks or errors. Users must use context managers or call close() explicitly.

Source code in affinity/client.py

def __del__(self) -> None:
    """Warn if client was not properly closed.

    NOTE: We intentionally do NOT attempt async cleanup here.
    asyncio.create_task() in __del__ is unsafe - the task may be garbage
    collected before completion, leading to resource leaks or errors.
    Users must use context managers or call close() explicitly.
    """
    # Use getattr to handle case where __init__ failed before setting _closed
    if not getattr(self, "_closed", True) and not getattr(self, "_entered_context", True):
        warnings.warn(
            "AsyncAffinity client was not closed. "
            "Use 'async with AsyncAffinity.from_env() as client:' "
            "or call await client.close() when done.",
            ResourceWarning,
            stacklevel=2,
        )

__init__(api_key: str, *, v1_base_url: str = V1_BASE_URL, v2_base_url: str = V2_BASE_URL, v1_auth_mode: Literal['bearer', 'basic'] = 'bearer', transport: httpx.BaseTransport | None = None, async_transport: httpx.AsyncBaseTransport | None = None, enable_beta_endpoints: bool = False, allow_insecure_download_redirects: bool = False, expected_v2_version: str | None = None, affinity_api_version: str | None = None, timeout: float = 30.0, max_retries: int = 3, enable_cache: bool = False, cache_ttl: float = 300.0, log_requests: bool = False, on_request: RequestHook | None = None, on_response: ResponseHook | None = None, on_error: ErrorHook | None = None, on_event: AnyEventHook | None = None, hook_error_policy: Literal['swallow', 'raise'] = 'swallow', policies: Policies | None = None)

Initialize the async Affinity client.

Parameters:

Name	Type	Description	Default
api_key	str	Your Affinity API key	required
v1_base_url	str	V1 API base URL (default: https://api.affinity.co)	V1_BASE_URL
v2_base_url	str	V2 API base URL (default: https://api.affinity.co/v2)	V2_BASE_URL
v1_auth_mode	Literal['bearer', 'basic']	Auth mode for V1 API ("bearer" or "basic")	'bearer'
transport	BaseTransport | None	Optional httpx transport (advanced; useful for mocking in tests)	None
async_transport	AsyncBaseTransport | None	Optional async httpx transport (advanced; useful for mocking in tests)	None
enable_beta_endpoints	bool	Enable beta V2 endpoints	False
allow_insecure_download_redirects	bool	Allow http:// redirects for file downloads. Not recommended; prefer HTTPS-only downloads.	False
expected_v2_version	str | None	Deprecated. Expected V2 API version, used only to label diagnostics (a mismatch with the echoed version is logged once). Use affinity_api_version to pin the version instead.	None
affinity_api_version	str | None	Affinity V2 API version to request, sent as the X-Affinity-Api-Version header on every V2 request (e.g. "2026-09-17", or "current" for the newest). None (default), "auto" or "key-default" send no header, so the API key's default version applies. V1 requests never carry it. Invalid values raise ConfigurationError; a version Affinity rejects raises UnsupportedApiVersionError.	None
timeout	float	Request timeout in seconds	30.0
max_retries	int	Maximum retries for rate-limited requests	3
enable_cache	bool	Enable response caching for field metadata	False
cache_ttl	float	Cache TTL in seconds	300.0
log_requests	bool	Log all HTTP requests (for debugging)	False
on_request	RequestHook | None	Hook called before each request (DX-008)	None
on_response	ResponseHook | None	Hook called after each response (DX-008)	None
on_error	ErrorHook | None	Hook called when a request raises (DX-008)	None
on_event	AnyEventHook | None	Event hook called for request/response lifecycle events (DX-008)	None
hook_error_policy	Literal['swallow', 'raise']	What to do if hooks raise ("swallow" or "raise")	'swallow'

Source code in affinity/client.py

def __init__(
    self,
    api_key: str,
    *,
    v1_base_url: str = V1_BASE_URL,
    v2_base_url: str = V2_BASE_URL,
    v1_auth_mode: Literal["bearer", "basic"] = "bearer",
    transport: httpx.BaseTransport | None = None,
    async_transport: httpx.AsyncBaseTransport | None = None,
    enable_beta_endpoints: bool = False,
    allow_insecure_download_redirects: bool = False,
    expected_v2_version: str | None = None,
    affinity_api_version: str | None = None,
    timeout: float = 30.0,
    max_retries: int = 3,
    enable_cache: bool = False,
    cache_ttl: float = 300.0,
    log_requests: bool = False,
    on_request: RequestHook | None = None,
    on_response: ResponseHook | None = None,
    on_error: ErrorHook | None = None,
    on_event: AnyEventHook | None = None,
    hook_error_policy: Literal["swallow", "raise"] = "swallow",
    policies: Policies | None = None,
):
    """
    Initialize the async Affinity client.

    Args:
        api_key: Your Affinity API key
        v1_base_url: V1 API base URL (default: https://api.affinity.co)
        v2_base_url: V2 API base URL (default: https://api.affinity.co/v2)
        v1_auth_mode: Auth mode for V1 API ("bearer" or "basic")
        transport: Optional `httpx` transport (advanced; useful for mocking in tests)
        async_transport: Optional async `httpx` transport (advanced; useful for mocking in
            tests)
        enable_beta_endpoints: Enable beta V2 endpoints
        allow_insecure_download_redirects: Allow `http://` redirects for file downloads.
            Not recommended; prefer HTTPS-only downloads.
        expected_v2_version: Deprecated. Expected V2 API version, used only to label
            diagnostics (a mismatch with the echoed version is logged once). Use
            ``affinity_api_version`` to pin the version instead.
        affinity_api_version: Affinity V2 API version to request, sent as the
            ``X-Affinity-Api-Version`` header on every V2 request (e.g. "2026-09-17",
            or "current" for the newest). None (default), "auto" or "key-default"
            send no header, so the API key's default version applies. V1 requests
            never carry it. Invalid values raise ``ConfigurationError``; a version
            Affinity rejects raises ``UnsupportedApiVersionError``.
        timeout: Request timeout in seconds
        max_retries: Maximum retries for rate-limited requests
        enable_cache: Enable response caching for field metadata
        cache_ttl: Cache TTL in seconds
        log_requests: Log all HTTP requests (for debugging)
        on_request: Hook called before each request (DX-008)
        on_response: Hook called after each response (DX-008)
        on_error: Hook called when a request raises (DX-008)
        on_event: Event hook called for request/response lifecycle events (DX-008)
        hook_error_policy: What to do if hooks raise ("swallow" or "raise")
    """
    config = ClientConfig(
        api_key=api_key,
        v1_base_url=v1_base_url,
        v2_base_url=v2_base_url,
        v1_auth_mode=v1_auth_mode,
        transport=transport,
        async_transport=async_transport,
        enable_beta_endpoints=enable_beta_endpoints,
        allow_insecure_download_redirects=allow_insecure_download_redirects,
        expected_v2_version=expected_v2_version,
        affinity_api_version=affinity_api_version,
        timeout=timeout,
        max_retries=max_retries,
        enable_cache=enable_cache,
        cache_ttl=cache_ttl,
        log_requests=log_requests,
        on_request=on_request,
        on_response=on_response,
        on_error=on_error,
        on_event=on_event,
        hook_error_policy=hook_error_policy,
        policies=policies or Policies(),
    )
    self._http = AsyncHTTPClient(config)

    # Resource management tracking
    self._closed = False
    self._entered_context = False

    self._companies: AsyncCompanyService | None = None
    self._persons: AsyncPersonService | None = None
    self._opportunities: AsyncOpportunityService | None = None
    self._lists: AsyncListService | None = None
    self._tasks: AsyncTaskService | None = None
    self._notes: AsyncNoteService | None = None
    self._transcripts: AsyncTranscriptService | None = None
    self._reminders: AsyncReminderService | None = None
    self._webhooks: AsyncWebhookService | None = None
    self._teams: AsyncTeamService | None = None
    self._users: AsyncUserService | None = None
    self._feedback: AsyncFeedbackService | None = None
    self._interactions: AsyncInteractionService | None = None
    self._fields: AsyncFieldService | None = None
    self._field_values: AsyncFieldValueService | None = None
    self._field_value_changes: AsyncFieldValueChangesService | None = None
    self._files: AsyncEntityFileService | None = None
    self._relationships: AsyncRelationshipStrengthService | None = None
    self._auth: AsyncAuthService | None = None
    self._rate_limits: AsyncRateLimitService | None = None

clear_cache() -> None

Clear the response cache.

Source code in affinity/client.py

def clear_cache(self) -> None:
    """Clear the response cache."""
    if self._http.cache:
        self._http.cache.clear()

close() -> None async

Close the HTTP client.

Source code in affinity/client.py

async def close(self) -> None:
    """Close the HTTP client."""
    if not self._closed:
        await self._http.close()
        self._closed = True

from_env(*, env_var: str = _DEFAULT_API_KEY_ENV_VAR, load_dotenv: bool = False, dotenv_path: str | os.PathLike[str] | None = None, dotenv_override: bool = False, transport: httpx.BaseTransport | None = None, async_transport: httpx.AsyncBaseTransport | None = None, policies: Policies | None = None, on_event: AnyEventHook | None = None, hook_error_policy: Literal['swallow', 'raise'] = 'swallow', **kwargs: Any) -> AsyncAffinity classmethod

Create an async client using an API key from the environment.

By default, reads AFFINITY_API_KEY. For local development, you can optionally load a .env file (requires python-dotenv).

Source code in affinity/client.py

@classmethod
def from_env(
    cls,
    *,
    env_var: str = _DEFAULT_API_KEY_ENV_VAR,
    load_dotenv: bool = False,
    dotenv_path: str | os.PathLike[str] | None = None,
    dotenv_override: bool = False,
    transport: httpx.BaseTransport | None = None,
    async_transport: httpx.AsyncBaseTransport | None = None,
    policies: Policies | None = None,
    on_event: AnyEventHook | None = None,
    hook_error_policy: Literal["swallow", "raise"] = "swallow",
    **kwargs: Any,
) -> AsyncAffinity:
    """
    Create an async client using an API key from the environment.

    By default, reads `AFFINITY_API_KEY`. For local development, you can optionally
    load a `.env` file (requires `python-dotenv`).
    """
    api_key = _api_key_from_env(
        env_var=env_var,
        load_dotenv=load_dotenv,
        dotenv_path=dotenv_path,
        dotenv_override=dotenv_override,
    )
    return cls(
        api_key=api_key,
        transport=transport,
        async_transport=async_transport,
        policies=policies,
        on_event=on_event,
        hook_error_policy=hook_error_policy,
        **kwargs,
    )

get_key_default_api_version() -> str | None async

The API key's default V2 API version (one unversioned V2 request).

Source code in affinity/client.py

async def get_key_default_api_version(self) -> str | None:
    """The API key's default V2 API version (one unversioned V2 request)."""
    return await self._http.probe_key_default_api_version()

read_only(api_key: str, **kwargs: Any) -> AsyncAffinity classmethod

Create a read-only async client that blocks all write operations.

See Affinity.read_only() for details.

Source code in affinity/client.py

@classmethod
def read_only(
    cls,
    api_key: str,
    **kwargs: Any,
) -> AsyncAffinity:
    """
    Create a read-only async client that blocks all write operations.

    See Affinity.read_only() for details.
    """
    if "policies" in kwargs:
        raise ValueError(
            "Cannot specify 'policies' with read_only(); "
            "use AsyncAffinity() constructor directly for custom policies"
        )
    return cls(
        api_key=api_key,
        policies=Policies(write=WritePolicy.DENY),
        **kwargs,
    )

read_only_from_env(*, env_var: str = _DEFAULT_API_KEY_ENV_VAR, **kwargs: Any) -> AsyncAffinity classmethod

Create a read-only async client using API key from environment.

See Affinity.read_only_from_env() for details.

Source code in affinity/client.py

@classmethod
def read_only_from_env(
    cls,
    *,
    env_var: str = _DEFAULT_API_KEY_ENV_VAR,
    **kwargs: Any,
) -> AsyncAffinity:
    """
    Create a read-only async client using API key from environment.

    See Affinity.read_only_from_env() for details.
    """
    if "policies" in kwargs:
        raise ValueError(
            "Cannot specify 'policies' with read_only_from_env(); "
            "use AsyncAffinity.from_env() directly for custom policies"
        )
    return cls.from_env(
        env_var=env_var,
        policies=Policies(write=WritePolicy.DENY),
        **kwargs,
    )

require_api_version(minimum: str, *, operation: str) -> None

Raise ApiVersionTooOldError if this client is pinned to a version older than minimum (nothing is sent). Unpinned clients always pass: version-gated calls send their minimum version themselves.

Source code in affinity/client.py

def require_api_version(self, minimum: str, *, operation: str) -> None:
    """
    Raise ``ApiVersionTooOldError`` if this client is pinned to a version older than
    ``minimum`` (nothing is sent). Unpinned clients always pass: version-gated calls send
    their minimum version themselves.
    """
    self._http.require_api_version(minimum, operation=operation)

whoami() -> WhoAmI async

Convenience wrapper for client.auth.whoami().

Source code in affinity/client.py

async def whoami(self) -> WhoAmI:
    """Convenience wrapper for `client.auth.whoami()`."""
    return await self.auth.whoami()