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

1from datetime import datetime 1 ctx1a

2from typing import Annotated 1 ctx1a

3 

4from fastapi import Path 1 ctx1a

5from pydantic import UUID4, Field 1 ctx1a

6 

7from polar.kit.metadata import ( 1 ctx1a

8 MetadataInputMixin, 

9 MetadataOutputMixin, 

10) 

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

12from polar.meter.aggregation import Aggregation 1 ctx1a

13from polar.meter.filter import Filter 1 ctx1a

14from polar.organization.schemas import OrganizationID 1 ctx1a

15 

16NAME_DESCRIPTION = ( 1 ctx1a

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

18) 

19_filter_description = ( 1 ctx1a

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

21) 

22_aggregation_description = ( 1 ctx1a

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

24) 

25 

26 

27class MeterCreate(Schema, MetadataInputMixin): 1 ctx1a

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

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

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

31 organization_id: OrganizationID | None = Field( 1 ctx1a

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): 1 ctx1a

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

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

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

44 is_archived: bool | None = Field( 1 ctx1a

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): 1 ctx1a

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

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

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

57 organization_id: UUID4 = Field( 1 ctx1a

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

59 ) 

60 archived_at: datetime | None = Field( 1 ctx1a

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.")] 1 ctx1a

66 

67 

68class MeterQuantity(Schema): 1 ctx1a

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

70 quantity: float = Field( 1 ctx1a

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

72 ) 

73 

74 

75class MeterQuantities(Schema): 1 ctx1a

76 quantities: list[MeterQuantity] 1 ctx1a

77 total: float = Field( 1 ctx1a

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

79 )