Coverage for polar/meter/schemas.py: 100%

1from datetime import datetime 

2from typing import Annotated 

3 

4from fastapi import Path 

5from pydantic import UUID4, Field 

6 

7from polar.kit.metadata import ( 

8 MetadataInputMixin, 

9 MetadataOutputMixin, 

10) 

11from polar.kit.schemas import IDSchema, Schema, TimestampedSchema 

12from polar.meter.aggregation import Aggregation 

13from polar.meter.filter import Filter 

14from polar.organization.schemas import OrganizationID 

15 

16NAME_DESCRIPTION = ( 

17 "The name of the meter. Will be shown on customer's invoices and usage." 

18) 

19_filter_description = ( 

20 "The filter to apply on events that'll be used to calculate the meter." 

21) 

22_aggregation_description = ( 

23 "The aggregation to apply on the filtered events to calculate the meter." 

24) 

25 

26 

27class MeterCreate(Schema, MetadataInputMixin): 

28 name: str = Field(..., description=NAME_DESCRIPTION, min_length=3) 

29 filter: Filter = Field(..., description=_filter_description) 

30 aggregation: Aggregation = Field(..., description=_aggregation_description) 

31 organization_id: OrganizationID | None = Field( 

32 default=None, 

33 description=( 

34 "The ID of the organization owning the meter. " 

35 "**Required unless you use an organization token.**" 

36 ), 

37 ) 

38 

39 

40class MeterUpdate(Schema, MetadataInputMixin): 

41 name: str | None = Field(None, description=NAME_DESCRIPTION, min_length=3) 

42 filter: Filter | None = Field(None, description=_filter_description) 

43 aggregation: Aggregation | None = Field(None, description=_aggregation_description) 

44 is_archived: bool | None = Field( 

45 None, 

46 description=( 

47 "Whether the meter is archived. " 

48 "Archived meters are no longer used for billing." 

49 ), 

50 ) 

51 

52 

53class Meter(IDSchema, TimestampedSchema, MetadataOutputMixin): 

54 name: str = Field(..., description=NAME_DESCRIPTION) 

55 filter: Filter = Field(..., description=_filter_description) 

56 aggregation: Aggregation = Field(..., description=_aggregation_description) 

57 organization_id: UUID4 = Field( 

58 ..., description="The ID of the organization owning the meter." 

59 ) 

60 archived_at: datetime | None = Field( 

61 None, description="Whether the meter is archived and the time it was archived." 

62 ) 

63 

64 

65MeterID = Annotated[UUID4, Path(description="The meter ID.")] 

66 

67 

68class MeterQuantity(Schema): 

69 timestamp: datetime = Field(description="The timestamp for the current period.") 

70 quantity: float = Field( 

71 description="The quantity for the current period.", examples=[10.0] 

72 ) 

73 

74 

75class MeterQuantities(Schema): 

76 quantities: list[MeterQuantity] 

77 total: float = Field( 

78 description="The total quantity for the period.", examples=[100.0] 

79 )