Skip to content

Reminders

Reminders (V2, with two V1 exceptions; see the module docstring).

Source code in affinity/services/reminders.py 
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
class ReminderService:
    """Reminders (V2, with two V1 exceptions; see the module docstring)."""

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

    def list(
        self,
        *,
        person_id: PersonId | None = None,
        company_id: CompanyId | None = None,
        opportunity_id: OpportunityId | None = None,
        creator_id: UserId | None = None,
        owner_id: UserId | None = None,
        completer_id: UserId | None = None,
        type: ReminderType | None = None,
        reset_type: ReminderResetType | None = None,
        status: ReminderStatus | None = None,
        due_after: datetime | None = None,
        due_before: datetime | None = None,
        ids: Sequence[ReminderIdType] | None = None,
        created_after: datetime | None = None,
        created_before: datetime | None = None,
        updated_after: datetime | None = None,
        updated_before: datetime | None = None,
        limit: int | None = None,
        total_count: bool = False,
        cursor: str | None = None,
    ) -> PaginatedResponse[Reminder]:
        """One page of the reminders visible to the caller, soonest due first.

        ``*_after`` bounds are inclusive and ``*_before`` exclusive (whole seconds). ``limit``
        is 1-100. Pass ``cursor`` (``pagination.next_cursor``) alone for the next page;
        :meth:`iter` follows every page.
        """
        ids = materialize_ids(ids)
        filters: dict[str, Any] = {
            "ids": ids,
            "person_id": person_id,
            "company_id": company_id,
            "opportunity_id": opportunity_id,
            "creator_id": creator_id,
            "owner_id": owner_id,
            "completer_id": completer_id,
            "type": type,
            "reset_type": reset_type,
            "status": status,
            "due_after": due_after,
            "due_before": due_before,
            "created_after": created_after,
            "created_before": created_before,
            "updated_after": updated_after,
            "updated_before": updated_before,
        }
        check_cursor_alone(cursor, *filters.values(), limit, total_count)
        if cursor is not None:
            return page_of(Reminder, self._client.get_url(cursor))
        params = reminder_params(**filters, limit=limit, total_count=total_count)
        if no_ids(ids):
            return empty_page(Reminder, total_count=total_count)
        return page_of(Reminder, self._client.get("/reminders", params=params or None))

    def iter(
        self,
        *,
        person_id: PersonId | None = None,
        company_id: CompanyId | None = None,
        opportunity_id: OpportunityId | None = None,
        creator_id: UserId | None = None,
        owner_id: UserId | None = None,
        completer_id: UserId | None = None,
        type: ReminderType | None = None,
        reset_type: ReminderResetType | None = None,
        status: ReminderStatus | None = None,
        due_after: datetime | None = None,
        due_before: datetime | None = None,
        ids: Sequence[ReminderIdType] | None = None,
        created_after: datetime | None = None,
        created_before: datetime | None = None,
        updated_after: datetime | None = None,
        updated_before: datetime | None = None,
        limit: int | None = None,
    ) -> Iterator[Reminder]:
        """All reminders matching the filters (every page); see :meth:`list`."""
        kwargs = locals().copy()
        kwargs.pop("self")

        def fetch_page(next_url: str | None) -> PaginatedResponse[Reminder]:
            if next_url:
                return self.list(cursor=next_url)
            return self.list(**kwargs)

        return PageIterator(fetch_page)

    def list_batch(
        self,
        *,
        person_ids: Sequence[PersonId] | None = None,
        company_ids: Sequence[CompanyId] | None = None,
        opportunity_ids: Sequence[OpportunityId] | None = None,
        status: ReminderStatus | None = None,
        type: ReminderType | None = None,
        reset_type: ReminderResetType | None = None,
        creator_id: UserId | None = None,
        owner_id: UserId | None = None,
        due_before: datetime | None = None,
        due_after: datetime | None = None,
    ) -> dict[int, builtins.list[Reminder]]:
        """Reminders for many entities of one kind: ``{entity_id: [reminders]}``.

        One request per 50 ids (an OR filter), plus further pages. Every requested id is a key;
        entities without matching reminders map to an empty list.
        """
        field, attr, unique = _batch_target(person_ids, company_ids, opportunity_ids)
        results: dict[int, builtins.list[Reminder]] = {i: [] for i in unique}
        filters: dict[str, Any] = {
            "status": status,
            "type": type,
            "reset_type": reset_type,
            "creator_id": creator_id,
            "owner_id": owner_id,
            "due_before": due_before,
            "due_after": due_after,
        }
        for start in range(0, len(unique), BATCH_CHUNK):
            chunk = unique[start : start + BATCH_CHUNK]
            params = reminder_params(**filters, either=(field, chunk), limit=100)
            page = page_of(Reminder, self._client.get("/reminders", params=params))
            while True:
                for reminder in page.data:
                    owner = _owner_of(reminder, attr)
                    if owner in results:
                        results[owner].append(reminder)
                if not page.next_cursor:
                    break
                page = page_of(Reminder, self._client.get_url(page.next_cursor))
        return results

    def get(self, reminder_id: ReminderIdType) -> Reminder:
        """One reminder (``NotFoundError`` when it doesn't exist or isn't visible)."""
        return Reminder.model_validate(
            self._client.get(
                f"/reminders/{int(reminder_id)}", min_api_version=GET_DELETE_MIN_API_VERSION
            )
        )

    def create(self, data: ReminderCreate) -> Reminder:
        """Create a reminder attached to one person, company or opportunity; the caller becomes
        its creator."""
        _check_create(data)
        return Reminder.model_validate(self._client.post("/reminders", json=create_body(data)))

    def update(self, reminder_id: ReminderIdType, data: ReminderUpdate) -> Reminder:
        """Change a reminder (only what is set in ``data``) and return it as read back.

        Two requests on V2 (the update has no response body). With ``type`` set, the whole update
        is one V1 request. If V2 refuses (only the creator or owner may update there), the update
        is retried once through V1.
        """
        if data.type is not None:
            return reminder_from_v1(
                self._client.put(
                    f"/reminders/{int(reminder_id)}", json=v1_update_body(data), v1=True
                )
            )
        body = update_body(data)
        try:
            self._client.post(
                f"/reminders/{int(reminder_id)}", json=body, min_api_version=UPDATE_MIN_API_VERSION
            )
        except AuthorizationError as refused:
            # V2 lets only the creator or owner change a reminder; V1 states no such rule
            if data.clear_content:
                raise
            try:
                body = v1_retry_body(data, self.get(reminder_id))
                result = self._client.put(f"/reminders/{int(reminder_id)}", json=body, v1=True)
            except AffinityError as retry_error:
                raise refused from retry_error
            return reminder_from_v1(result)
        return self.get(reminder_id)

    def delete(self, reminder_id: ReminderIdType) -> bool:
        """Delete a reminder. Returns True.

        If V2 refuses (only the creator or owner may delete there), retried once through V1.
        """
        try:
            self._client.delete(
                f"/reminders/{int(reminder_id)}", min_api_version=GET_DELETE_MIN_API_VERSION
            )
        except AuthorizationError as refused:
            try:
                self._client.delete(f"/reminders/{int(reminder_id)}", v1=True)
            except AffinityError as retry_error:
                raise refused from retry_error
        return True

create(data: ReminderCreate) -> Reminder

Create a reminder attached to one person, company or opportunity; the caller becomes its creator.

Source code in affinity/services/reminders.py 
474
475
476
477
478
def create(self, data: ReminderCreate) -> Reminder:
    """Create a reminder attached to one person, company or opportunity; the caller becomes
    its creator."""
    _check_create(data)
    return Reminder.model_validate(self._client.post("/reminders", json=create_body(data)))

delete(reminder_id: ReminderIdType) -> bool

Delete a reminder. Returns True.

If V2 refuses (only the creator or owner may delete there), retried once through V1.

Source code in affinity/services/reminders.py 
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
def delete(self, reminder_id: ReminderIdType) -> bool:
    """Delete a reminder. Returns True.

    If V2 refuses (only the creator or owner may delete there), retried once through V1.
    """
    try:
        self._client.delete(
            f"/reminders/{int(reminder_id)}", min_api_version=GET_DELETE_MIN_API_VERSION
        )
    except AuthorizationError as refused:
        try:
            self._client.delete(f"/reminders/{int(reminder_id)}", v1=True)
        except AffinityError as retry_error:
            raise refused from retry_error
    return True

get(reminder_id: ReminderIdType) -> Reminder

One reminder (NotFoundError when it doesn't exist or isn't visible).

Source code in affinity/services/reminders.py 
466
467
468
469
470
471
472
def get(self, reminder_id: ReminderIdType) -> Reminder:
    """One reminder (``NotFoundError`` when it doesn't exist or isn't visible)."""
    return Reminder.model_validate(
        self._client.get(
            f"/reminders/{int(reminder_id)}", min_api_version=GET_DELETE_MIN_API_VERSION
        )
    )

iter(*, person_id: PersonId | None = None, company_id: CompanyId | None = None, opportunity_id: OpportunityId | None = None, creator_id: UserId | None = None, owner_id: UserId | None = None, completer_id: UserId | None = None, type: ReminderType | None = None, reset_type: ReminderResetType | None = None, status: ReminderStatus | None = None, due_after: datetime | None = None, due_before: datetime | None = None, ids: Sequence[ReminderIdType] | None = None, created_after: datetime | None = None, created_before: datetime | None = None, updated_after: datetime | None = None, updated_before: datetime | None = None, limit: int | None = None) -> Iterator[Reminder]

All reminders matching the filters (every page); see :meth:list.

Source code in affinity/services/reminders.py 
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
def iter(
    self,
    *,
    person_id: PersonId | None = None,
    company_id: CompanyId | None = None,
    opportunity_id: OpportunityId | None = None,
    creator_id: UserId | None = None,
    owner_id: UserId | None = None,
    completer_id: UserId | None = None,
    type: ReminderType | None = None,
    reset_type: ReminderResetType | None = None,
    status: ReminderStatus | None = None,
    due_after: datetime | None = None,
    due_before: datetime | None = None,
    ids: Sequence[ReminderIdType] | None = None,
    created_after: datetime | None = None,
    created_before: datetime | None = None,
    updated_after: datetime | None = None,
    updated_before: datetime | None = None,
    limit: int | None = None,
) -> Iterator[Reminder]:
    """All reminders matching the filters (every page); see :meth:`list`."""
    kwargs = locals().copy()
    kwargs.pop("self")

    def fetch_page(next_url: str | None) -> PaginatedResponse[Reminder]:
        if next_url:
            return self.list(cursor=next_url)
        return self.list(**kwargs)

    return PageIterator(fetch_page)

list(*, person_id: PersonId | None = None, company_id: CompanyId | None = None, opportunity_id: OpportunityId | None = None, creator_id: UserId | None = None, owner_id: UserId | None = None, completer_id: UserId | None = None, type: ReminderType | None = None, reset_type: ReminderResetType | None = None, status: ReminderStatus | None = None, due_after: datetime | None = None, due_before: datetime | None = None, ids: Sequence[ReminderIdType] | None = None, created_after: datetime | None = None, created_before: datetime | None = None, updated_after: datetime | None = None, updated_before: datetime | None = None, limit: int | None = None, total_count: bool = False, cursor: str | None = None) -> PaginatedResponse[Reminder]

One page of the reminders visible to the caller, soonest due first.

*_after bounds are inclusive and *_before exclusive (whole seconds). limit is 1-100. Pass cursor (pagination.next_cursor) alone for the next page; :meth:iter follows every page.

Source code in affinity/services/reminders.py 
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
def list(
    self,
    *,
    person_id: PersonId | None = None,
    company_id: CompanyId | None = None,
    opportunity_id: OpportunityId | None = None,
    creator_id: UserId | None = None,
    owner_id: UserId | None = None,
    completer_id: UserId | None = None,
    type: ReminderType | None = None,
    reset_type: ReminderResetType | None = None,
    status: ReminderStatus | None = None,
    due_after: datetime | None = None,
    due_before: datetime | None = None,
    ids: Sequence[ReminderIdType] | None = None,
    created_after: datetime | None = None,
    created_before: datetime | None = None,
    updated_after: datetime | None = None,
    updated_before: datetime | None = None,
    limit: int | None = None,
    total_count: bool = False,
    cursor: str | None = None,
) -> PaginatedResponse[Reminder]:
    """One page of the reminders visible to the caller, soonest due first.

    ``*_after`` bounds are inclusive and ``*_before`` exclusive (whole seconds). ``limit``
    is 1-100. Pass ``cursor`` (``pagination.next_cursor``) alone for the next page;
    :meth:`iter` follows every page.
    """
    ids = materialize_ids(ids)
    filters: dict[str, Any] = {
        "ids": ids,
        "person_id": person_id,
        "company_id": company_id,
        "opportunity_id": opportunity_id,
        "creator_id": creator_id,
        "owner_id": owner_id,
        "completer_id": completer_id,
        "type": type,
        "reset_type": reset_type,
        "status": status,
        "due_after": due_after,
        "due_before": due_before,
        "created_after": created_after,
        "created_before": created_before,
        "updated_after": updated_after,
        "updated_before": updated_before,
    }
    check_cursor_alone(cursor, *filters.values(), limit, total_count)
    if cursor is not None:
        return page_of(Reminder, self._client.get_url(cursor))
    params = reminder_params(**filters, limit=limit, total_count=total_count)
    if no_ids(ids):
        return empty_page(Reminder, total_count=total_count)
    return page_of(Reminder, self._client.get("/reminders", params=params or None))

list_batch(*, person_ids: Sequence[PersonId] | None = None, company_ids: Sequence[CompanyId] | None = None, opportunity_ids: Sequence[OpportunityId] | None = None, status: ReminderStatus | None = None, type: ReminderType | None = None, reset_type: ReminderResetType | None = None, creator_id: UserId | None = None, owner_id: UserId | None = None, due_before: datetime | None = None, due_after: datetime | None = None) -> dict[int, builtins.list[Reminder]]

Reminders for many entities of one kind: {entity_id: [reminders]}.

One request per 50 ids (an OR filter), plus further pages. Every requested id is a key; entities without matching reminders map to an empty list.

Source code in affinity/services/reminders.py 
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
def list_batch(
    self,
    *,
    person_ids: Sequence[PersonId] | None = None,
    company_ids: Sequence[CompanyId] | None = None,
    opportunity_ids: Sequence[OpportunityId] | None = None,
    status: ReminderStatus | None = None,
    type: ReminderType | None = None,
    reset_type: ReminderResetType | None = None,
    creator_id: UserId | None = None,
    owner_id: UserId | None = None,
    due_before: datetime | None = None,
    due_after: datetime | None = None,
) -> dict[int, builtins.list[Reminder]]:
    """Reminders for many entities of one kind: ``{entity_id: [reminders]}``.

    One request per 50 ids (an OR filter), plus further pages. Every requested id is a key;
    entities without matching reminders map to an empty list.
    """
    field, attr, unique = _batch_target(person_ids, company_ids, opportunity_ids)
    results: dict[int, builtins.list[Reminder]] = {i: [] for i in unique}
    filters: dict[str, Any] = {
        "status": status,
        "type": type,
        "reset_type": reset_type,
        "creator_id": creator_id,
        "owner_id": owner_id,
        "due_before": due_before,
        "due_after": due_after,
    }
    for start in range(0, len(unique), BATCH_CHUNK):
        chunk = unique[start : start + BATCH_CHUNK]
        params = reminder_params(**filters, either=(field, chunk), limit=100)
        page = page_of(Reminder, self._client.get("/reminders", params=params))
        while True:
            for reminder in page.data:
                owner = _owner_of(reminder, attr)
                if owner in results:
                    results[owner].append(reminder)
            if not page.next_cursor:
                break
            page = page_of(Reminder, self._client.get_url(page.next_cursor))
    return results

update(reminder_id: ReminderIdType, data: ReminderUpdate) -> Reminder

Change a reminder (only what is set in data) and return it as read back.

Two requests on V2 (the update has no response body). With type set, the whole update is one V1 request. If V2 refuses (only the creator or owner may update there), the update is retried once through V1.

Source code in affinity/services/reminders.py 
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
def update(self, reminder_id: ReminderIdType, data: ReminderUpdate) -> Reminder:
    """Change a reminder (only what is set in ``data``) and return it as read back.

    Two requests on V2 (the update has no response body). With ``type`` set, the whole update
    is one V1 request. If V2 refuses (only the creator or owner may update there), the update
    is retried once through V1.
    """
    if data.type is not None:
        return reminder_from_v1(
            self._client.put(
                f"/reminders/{int(reminder_id)}", json=v1_update_body(data), v1=True
            )
        )
    body = update_body(data)
    try:
        self._client.post(
            f"/reminders/{int(reminder_id)}", json=body, min_api_version=UPDATE_MIN_API_VERSION
        )
    except AuthorizationError as refused:
        # V2 lets only the creator or owner change a reminder; V1 states no such rule
        if data.clear_content:
            raise
        try:
            body = v1_retry_body(data, self.get(reminder_id))
            result = self._client.put(f"/reminders/{int(reminder_id)}", json=body, v1=True)
        except AffinityError as retry_error:
            raise refused from retry_error
        return reminder_from_v1(result)
    return self.get(reminder_id)

Async reminders; same behavior as :class:ReminderService.

Source code in affinity/services/reminders.py 
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
class AsyncReminderService:
    """Async reminders; same behavior as :class:`ReminderService`."""

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

    async def list(
        self,
        *,
        person_id: PersonId | None = None,
        company_id: CompanyId | None = None,
        opportunity_id: OpportunityId | None = None,
        creator_id: UserId | None = None,
        owner_id: UserId | None = None,
        completer_id: UserId | None = None,
        type: ReminderType | None = None,
        reset_type: ReminderResetType | None = None,
        status: ReminderStatus | None = None,
        due_after: datetime | None = None,
        due_before: datetime | None = None,
        ids: Sequence[ReminderIdType] | None = None,
        created_after: datetime | None = None,
        created_before: datetime | None = None,
        updated_after: datetime | None = None,
        updated_before: datetime | None = None,
        limit: int | None = None,
        total_count: bool = False,
        cursor: str | None = None,
    ) -> PaginatedResponse[Reminder]:
        """One page of reminders; see :meth:`ReminderService.list`."""
        ids = materialize_ids(ids)
        filters: dict[str, Any] = {
            "ids": ids,
            "person_id": person_id,
            "company_id": company_id,
            "opportunity_id": opportunity_id,
            "creator_id": creator_id,
            "owner_id": owner_id,
            "completer_id": completer_id,
            "type": type,
            "reset_type": reset_type,
            "status": status,
            "due_after": due_after,
            "due_before": due_before,
            "created_after": created_after,
            "created_before": created_before,
            "updated_after": updated_after,
            "updated_before": updated_before,
        }
        check_cursor_alone(cursor, *filters.values(), limit, total_count)
        if cursor is not None:
            return page_of(Reminder, await self._client.get_url(cursor))
        params = reminder_params(**filters, limit=limit, total_count=total_count)
        if no_ids(ids):
            return empty_page(Reminder, total_count=total_count)
        return page_of(Reminder, await self._client.get("/reminders", params=params or None))

    def iter(
        self,
        *,
        person_id: PersonId | None = None,
        company_id: CompanyId | None = None,
        opportunity_id: OpportunityId | None = None,
        creator_id: UserId | None = None,
        owner_id: UserId | None = None,
        completer_id: UserId | None = None,
        type: ReminderType | None = None,
        reset_type: ReminderResetType | None = None,
        status: ReminderStatus | None = None,
        due_after: datetime | None = None,
        due_before: datetime | None = None,
        ids: Sequence[ReminderIdType] | None = None,
        created_after: datetime | None = None,
        created_before: datetime | None = None,
        updated_after: datetime | None = None,
        updated_before: datetime | None = None,
        limit: int | None = None,
    ) -> AsyncIterator[Reminder]:
        """All reminders matching the filters (every page); see :meth:`list`."""
        kwargs = locals().copy()
        kwargs.pop("self")

        async def fetch_page(next_url: str | None) -> PaginatedResponse[Reminder]:
            if next_url:
                return await self.list(cursor=next_url)
            return await self.list(**kwargs)

        return AsyncPageIterator(fetch_page)

    async def list_batch(
        self,
        *,
        person_ids: Sequence[PersonId] | None = None,
        company_ids: Sequence[CompanyId] | None = None,
        opportunity_ids: Sequence[OpportunityId] | None = None,
        status: ReminderStatus | None = None,
        type: ReminderType | None = None,
        reset_type: ReminderResetType | None = None,
        creator_id: UserId | None = None,
        owner_id: UserId | None = None,
        due_before: datetime | None = None,
        due_after: datetime | None = None,
        max_concurrent: int = 10,
        on_error: Literal["raise", "skip"] = "raise",
    ) -> dict[int, builtins.list[Reminder]]:
        """Reminders for many entities of one kind; see :meth:`ReminderService.list_batch`.

        Chunks of 50 ids run up to ``max_concurrent`` at a time. ``on_error="skip"`` doesn't
        raise for a failed chunk: its ids are left out of the result (they may have reminders
        the failed request didn't return), so every key present is complete and a missing
        requested id means its chunk failed.
        """
        field, attr, unique = _batch_target(person_ids, company_ids, opportunity_ids)
        if max_concurrent < 1:
            raise ValueError("max_concurrent must be at least 1")
        results: dict[int, builtins.list[Reminder]] = {i: [] for i in unique}
        filters: dict[str, Any] = {
            "status": status,
            "type": type,
            "reset_type": reset_type,
            "creator_id": creator_id,
            "owner_id": owner_id,
            "due_before": due_before,
            "due_after": due_after,
        }
        semaphore = asyncio.Semaphore(max_concurrent)

        async def fetch_chunk(chunk: builtins.list[int]) -> builtins.list[Reminder] | None:
            found: builtins.list[Reminder] = []
            async with semaphore:
                try:
                    params = reminder_params(**filters, either=(field, chunk), limit=100)
                    page = page_of(Reminder, await self._client.get("/reminders", params=params))
                    while True:
                        found.extend(page.data)
                        if not page.next_cursor:
                            break
                        page = page_of(Reminder, await self._client.get_url(page.next_cursor))
                except AffinityError:
                    if on_error == "raise":
                        raise
                    return None  # skip: the chunk's ids are left out, not reported as empty
            return found

        chunks = [unique[i : i + BATCH_CHUNK] for i in range(0, len(unique), BATCH_CHUNK)]
        fetched = await asyncio.gather(*(fetch_chunk(c) for c in chunks))
        for chunk, found in zip(chunks, fetched, strict=True):
            if found is None:
                for entity_id in chunk:
                    del results[entity_id]
                continue
            for reminder in found:
                owner = _owner_of(reminder, attr)
                if owner in results:
                    results[owner].append(reminder)
        return results

    async def get(self, reminder_id: ReminderIdType) -> Reminder:
        """One reminder; see :meth:`ReminderService.get`."""
        return Reminder.model_validate(
            await self._client.get(
                f"/reminders/{int(reminder_id)}", min_api_version=GET_DELETE_MIN_API_VERSION
            )
        )

    async def create(self, data: ReminderCreate) -> Reminder:
        """Create a reminder; see :meth:`ReminderService.create`."""
        _check_create(data)
        return Reminder.model_validate(
            await self._client.post("/reminders", json=create_body(data))
        )

    async def update(self, reminder_id: ReminderIdType, data: ReminderUpdate) -> Reminder:
        """Change a reminder and return it as read back; see :meth:`ReminderService.update`."""
        if data.type is not None:
            return reminder_from_v1(
                await self._client.put(
                    f"/reminders/{int(reminder_id)}", json=v1_update_body(data), v1=True
                )
            )
        body = update_body(data)
        try:
            await self._client.post(
                f"/reminders/{int(reminder_id)}", json=body, min_api_version=UPDATE_MIN_API_VERSION
            )
        except AuthorizationError as refused:
            if data.clear_content:
                raise
            try:
                body = v1_retry_body(data, await self.get(reminder_id))
                result = await self._client.put(
                    f"/reminders/{int(reminder_id)}", json=body, v1=True
                )
            except AffinityError as retry_error:
                raise refused from retry_error
            return reminder_from_v1(result)
        return await self.get(reminder_id)

    async def delete(self, reminder_id: ReminderIdType) -> bool:
        """Delete a reminder; see :meth:`ReminderService.delete`. Returns True."""
        try:
            await self._client.delete(
                f"/reminders/{int(reminder_id)}", min_api_version=GET_DELETE_MIN_API_VERSION
            )
        except AuthorizationError as refused:
            try:
                await self._client.delete(f"/reminders/{int(reminder_id)}", v1=True)
            except AffinityError as retry_error:
                raise refused from retry_error
        return True

create(data: ReminderCreate) -> Reminder async

Create a reminder; see :meth:ReminderService.create.

Source code in affinity/services/reminders.py 
692
693
694
695
696
697
async def create(self, data: ReminderCreate) -> Reminder:
    """Create a reminder; see :meth:`ReminderService.create`."""
    _check_create(data)
    return Reminder.model_validate(
        await self._client.post("/reminders", json=create_body(data))
    )

delete(reminder_id: ReminderIdType) -> bool async

Delete a reminder; see :meth:ReminderService.delete. Returns True.

Source code in affinity/services/reminders.py 
725
726
727
728
729
730
731
732
733
734
735
736
async def delete(self, reminder_id: ReminderIdType) -> bool:
    """Delete a reminder; see :meth:`ReminderService.delete`. Returns True."""
    try:
        await self._client.delete(
            f"/reminders/{int(reminder_id)}", min_api_version=GET_DELETE_MIN_API_VERSION
        )
    except AuthorizationError as refused:
        try:
            await self._client.delete(f"/reminders/{int(reminder_id)}", v1=True)
        except AffinityError as retry_error:
            raise refused from retry_error
    return True

get(reminder_id: ReminderIdType) -> Reminder async

One reminder; see :meth:ReminderService.get.

Source code in affinity/services/reminders.py 
684
685
686
687
688
689
690
async def get(self, reminder_id: ReminderIdType) -> Reminder:
    """One reminder; see :meth:`ReminderService.get`."""
    return Reminder.model_validate(
        await self._client.get(
            f"/reminders/{int(reminder_id)}", min_api_version=GET_DELETE_MIN_API_VERSION
        )
    )

iter(*, person_id: PersonId | None = None, company_id: CompanyId | None = None, opportunity_id: OpportunityId | None = None, creator_id: UserId | None = None, owner_id: UserId | None = None, completer_id: UserId | None = None, type: ReminderType | None = None, reset_type: ReminderResetType | None = None, status: ReminderStatus | None = None, due_after: datetime | None = None, due_before: datetime | None = None, ids: Sequence[ReminderIdType] | None = None, created_after: datetime | None = None, created_before: datetime | None = None, updated_after: datetime | None = None, updated_before: datetime | None = None, limit: int | None = None) -> AsyncIterator[Reminder]

All reminders matching the filters (every page); see :meth:list.

Source code in affinity/services/reminders.py 
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
def iter(
    self,
    *,
    person_id: PersonId | None = None,
    company_id: CompanyId | None = None,
    opportunity_id: OpportunityId | None = None,
    creator_id: UserId | None = None,
    owner_id: UserId | None = None,
    completer_id: UserId | None = None,
    type: ReminderType | None = None,
    reset_type: ReminderResetType | None = None,
    status: ReminderStatus | None = None,
    due_after: datetime | None = None,
    due_before: datetime | None = None,
    ids: Sequence[ReminderIdType] | None = None,
    created_after: datetime | None = None,
    created_before: datetime | None = None,
    updated_after: datetime | None = None,
    updated_before: datetime | None = None,
    limit: int | None = None,
) -> AsyncIterator[Reminder]:
    """All reminders matching the filters (every page); see :meth:`list`."""
    kwargs = locals().copy()
    kwargs.pop("self")

    async def fetch_page(next_url: str | None) -> PaginatedResponse[Reminder]:
        if next_url:
            return await self.list(cursor=next_url)
        return await self.list(**kwargs)

    return AsyncPageIterator(fetch_page)

list(*, person_id: PersonId | None = None, company_id: CompanyId | None = None, opportunity_id: OpportunityId | None = None, creator_id: UserId | None = None, owner_id: UserId | None = None, completer_id: UserId | None = None, type: ReminderType | None = None, reset_type: ReminderResetType | None = None, status: ReminderStatus | None = None, due_after: datetime | None = None, due_before: datetime | None = None, ids: Sequence[ReminderIdType] | None = None, created_after: datetime | None = None, created_before: datetime | None = None, updated_after: datetime | None = None, updated_before: datetime | None = None, limit: int | None = None, total_count: bool = False, cursor: str | None = None) -> PaginatedResponse[Reminder] async

One page of reminders; see :meth:ReminderService.list.

Source code in affinity/services/reminders.py 
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
async def list(
    self,
    *,
    person_id: PersonId | None = None,
    company_id: CompanyId | None = None,
    opportunity_id: OpportunityId | None = None,
    creator_id: UserId | None = None,
    owner_id: UserId | None = None,
    completer_id: UserId | None = None,
    type: ReminderType | None = None,
    reset_type: ReminderResetType | None = None,
    status: ReminderStatus | None = None,
    due_after: datetime | None = None,
    due_before: datetime | None = None,
    ids: Sequence[ReminderIdType] | None = None,
    created_after: datetime | None = None,
    created_before: datetime | None = None,
    updated_after: datetime | None = None,
    updated_before: datetime | None = None,
    limit: int | None = None,
    total_count: bool = False,
    cursor: str | None = None,
) -> PaginatedResponse[Reminder]:
    """One page of reminders; see :meth:`ReminderService.list`."""
    ids = materialize_ids(ids)
    filters: dict[str, Any] = {
        "ids": ids,
        "person_id": person_id,
        "company_id": company_id,
        "opportunity_id": opportunity_id,
        "creator_id": creator_id,
        "owner_id": owner_id,
        "completer_id": completer_id,
        "type": type,
        "reset_type": reset_type,
        "status": status,
        "due_after": due_after,
        "due_before": due_before,
        "created_after": created_after,
        "created_before": created_before,
        "updated_after": updated_after,
        "updated_before": updated_before,
    }
    check_cursor_alone(cursor, *filters.values(), limit, total_count)
    if cursor is not None:
        return page_of(Reminder, await self._client.get_url(cursor))
    params = reminder_params(**filters, limit=limit, total_count=total_count)
    if no_ids(ids):
        return empty_page(Reminder, total_count=total_count)
    return page_of(Reminder, await self._client.get("/reminders", params=params or None))

list_batch(*, person_ids: Sequence[PersonId] | None = None, company_ids: Sequence[CompanyId] | None = None, opportunity_ids: Sequence[OpportunityId] | None = None, status: ReminderStatus | None = None, type: ReminderType | None = None, reset_type: ReminderResetType | None = None, creator_id: UserId | None = None, owner_id: UserId | None = None, due_before: datetime | None = None, due_after: datetime | None = None, max_concurrent: int = 10, on_error: Literal['raise', 'skip'] = 'raise') -> dict[int, builtins.list[Reminder]] async

Reminders for many entities of one kind; see :meth:ReminderService.list_batch.

Chunks of 50 ids run up to max_concurrent at a time. on_error="skip" doesn't raise for a failed chunk: its ids are left out of the result (they may have reminders the failed request didn't return), so every key present is complete and a missing requested id means its chunk failed.

Source code in affinity/services/reminders.py 
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
async def list_batch(
    self,
    *,
    person_ids: Sequence[PersonId] | None = None,
    company_ids: Sequence[CompanyId] | None = None,
    opportunity_ids: Sequence[OpportunityId] | None = None,
    status: ReminderStatus | None = None,
    type: ReminderType | None = None,
    reset_type: ReminderResetType | None = None,
    creator_id: UserId | None = None,
    owner_id: UserId | None = None,
    due_before: datetime | None = None,
    due_after: datetime | None = None,
    max_concurrent: int = 10,
    on_error: Literal["raise", "skip"] = "raise",
) -> dict[int, builtins.list[Reminder]]:
    """Reminders for many entities of one kind; see :meth:`ReminderService.list_batch`.

    Chunks of 50 ids run up to ``max_concurrent`` at a time. ``on_error="skip"`` doesn't
    raise for a failed chunk: its ids are left out of the result (they may have reminders
    the failed request didn't return), so every key present is complete and a missing
    requested id means its chunk failed.
    """
    field, attr, unique = _batch_target(person_ids, company_ids, opportunity_ids)
    if max_concurrent < 1:
        raise ValueError("max_concurrent must be at least 1")
    results: dict[int, builtins.list[Reminder]] = {i: [] for i in unique}
    filters: dict[str, Any] = {
        "status": status,
        "type": type,
        "reset_type": reset_type,
        "creator_id": creator_id,
        "owner_id": owner_id,
        "due_before": due_before,
        "due_after": due_after,
    }
    semaphore = asyncio.Semaphore(max_concurrent)

    async def fetch_chunk(chunk: builtins.list[int]) -> builtins.list[Reminder] | None:
        found: builtins.list[Reminder] = []
        async with semaphore:
            try:
                params = reminder_params(**filters, either=(field, chunk), limit=100)
                page = page_of(Reminder, await self._client.get("/reminders", params=params))
                while True:
                    found.extend(page.data)
                    if not page.next_cursor:
                        break
                    page = page_of(Reminder, await self._client.get_url(page.next_cursor))
            except AffinityError:
                if on_error == "raise":
                    raise
                return None  # skip: the chunk's ids are left out, not reported as empty
        return found

    chunks = [unique[i : i + BATCH_CHUNK] for i in range(0, len(unique), BATCH_CHUNK)]
    fetched = await asyncio.gather(*(fetch_chunk(c) for c in chunks))
    for chunk, found in zip(chunks, fetched, strict=True):
        if found is None:
            for entity_id in chunk:
                del results[entity_id]
            continue
        for reminder in found:
            owner = _owner_of(reminder, attr)
            if owner in results:
                results[owner].append(reminder)
    return results

update(reminder_id: ReminderIdType, data: ReminderUpdate) -> Reminder async

Change a reminder and return it as read back; see :meth:ReminderService.update.

Source code in affinity/services/reminders.py 
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
async def update(self, reminder_id: ReminderIdType, data: ReminderUpdate) -> Reminder:
    """Change a reminder and return it as read back; see :meth:`ReminderService.update`."""
    if data.type is not None:
        return reminder_from_v1(
            await self._client.put(
                f"/reminders/{int(reminder_id)}", json=v1_update_body(data), v1=True
            )
        )
    body = update_body(data)
    try:
        await self._client.post(
            f"/reminders/{int(reminder_id)}", json=body, min_api_version=UPDATE_MIN_API_VERSION
        )
    except AuthorizationError as refused:
        if data.clear_content:
            raise
        try:
            body = v1_retry_body(data, await self.get(reminder_id))
            result = await self._client.put(
                f"/reminders/{int(reminder_id)}", json=body, v1=True
            )
        except AffinityError as retry_error:
            raise refused from retry_error
        return reminder_from_v1(result)
    return await self.get(reminder_id)