|
| 1 | +"""Bubble platform types for use with Pydantic models.""" |
| 2 | + |
| 3 | +from typing import Annotated, Any |
| 4 | + |
| 5 | +from pydantic import AfterValidator, BeforeValidator |
| 6 | + |
| 7 | +from bubble_data_api_client.exceptions import InvalidBubbleUIDError |
| 8 | +from bubble_data_api_client.validation import is_bubble_uid |
| 9 | + |
| 10 | + |
| 11 | +def _validate_bubble_uid(value: str) -> str: |
| 12 | + """Validate that a string is a valid Bubble UID.""" |
| 13 | + if not is_bubble_uid(value): |
| 14 | + raise InvalidBubbleUIDError(value) |
| 15 | + return value |
| 16 | + |
| 17 | + |
| 18 | +BubbleUID = Annotated[str, AfterValidator(_validate_bubble_uid)] |
| 19 | +"""A string validated as a Bubble UID (format: digits + 'x' + digits).""" |
| 20 | + |
| 21 | + |
| 22 | +def _coerce_optional_bubble_uid(value: Any) -> str | None: |
| 23 | + """Coerce to valid Bubble UID or None. Invalid values silently become None.""" |
| 24 | + if value is None or value == "": |
| 25 | + return None |
| 26 | + if not isinstance(value, str): |
| 27 | + return None |
| 28 | + if not is_bubble_uid(value): |
| 29 | + return None |
| 30 | + return value |
| 31 | + |
| 32 | + |
| 33 | +OptionalBubbleUID = Annotated[str | None, BeforeValidator(_coerce_optional_bubble_uid)] |
| 34 | +"""A Bubble UID that silently coerces invalid values (including empty string) to None.""" |
| 35 | + |
| 36 | + |
| 37 | +def _coerce_optional_bubble_uids(value: object) -> list[str] | None: |
| 38 | + """Coerce to list of valid Bubble UIDs or None. Empty/invalid becomes None.""" |
| 39 | + if not isinstance(value, list): |
| 40 | + return None |
| 41 | + result = [x for x in value if isinstance(x, str) and is_bubble_uid(x)] |
| 42 | + return result or None |
| 43 | + |
| 44 | + |
| 45 | +OptionalBubbleUIDs = Annotated[list[str] | None, BeforeValidator(_coerce_optional_bubble_uids)] |
| 46 | +"""A list of Bubble UIDs that silently coerces invalid/empty to None.""" |
0 commit comments