Coverage for polar/file/schemas.py: 90%
65 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-12-05 17:15 +0000
1from datetime import datetime 1a
2from typing import Annotated, Any, Literal, Self 1a
4from pydantic import UUID4, Discriminator, Field, TypeAdapter, computed_field 1a
6from polar.integrations.aws.s3.schemas import ( 1a
7 S3DownloadURL,
8 S3File,
9 S3FileCreate,
10 S3FileDownload,
11 S3FileUpload,
12 S3FileUploadCompleted,
13)
14from polar.kit.schemas import ClassName, MergeJSONSchema, Schema, SetSchemaReference 1a
15from polar.models.file import File, FileServiceTypes 1a
17from .s3 import S3_SERVICES 1a
20class FileCreateBase(S3FileCreate): 1a
21 service: FileServiceTypes 1a
22 version: str | None = None 1a
25class DownloadableFileCreate(FileCreateBase): 1a
26 """Schema to create a file to be associated with the downloadables benefit."""
28 service: Literal[FileServiceTypes.downloadable] 1a
31class ProductMediaFileCreate(FileCreateBase): 1a
32 """Schema to create a file to be used as a product media file."""
34 service: Literal[FileServiceTypes.product_media] 1a
35 mime_type: str = Field( 1a
36 description=(
37 "MIME type of the file. Only images are supported for this type of file."
38 ),
39 pattern=r"^image\/(jpeg|png|gif|webp|svg\+xml)$",
40 )
41 size: int = Field( 1a
42 description=(
43 "Size of the file. A maximum of 10 MB is allowed for this type of file."
44 ),
45 le=10 * 1024 * 1024,
46 )
49class OrganizationAvatarFileCreate(FileCreateBase): 1a
50 """Schema to create a file to be used as an organization avatar."""
52 service: Literal[FileServiceTypes.organization_avatar] 1a
53 mime_type: str = Field( 1a
54 description=(
55 "MIME type of the file. Only images are supported for this type of file."
56 ),
57 pattern=r"^image\/(jpeg|png|gif|webp|svg\+xml)$",
58 )
59 size: int = Field( 1a
60 description=(
61 "Size of the file. A maximum of 1 MB is allowed for this type of file."
62 ),
63 le=1 * 1024 * 1024,
64 )
67FileCreate = Annotated[ 1a
68 DownloadableFileCreate | ProductMediaFileCreate | OrganizationAvatarFileCreate,
69 Discriminator("service"),
70 SetSchemaReference("FileCreate"),
71]
74class FileReadBase(S3File): 1a
75 version: str | None 1a
76 service: FileServiceTypes 1a
77 is_uploaded: bool 1a
78 created_at: datetime 1a
81class DownloadableFileRead(FileReadBase): 1a
82 """File to be associated with the downloadables benefit."""
84 service: Literal[FileServiceTypes.downloadable] 1a
87class PublicFileReadBase(FileReadBase): 1a
88 @computed_field # type: ignore[prop-decorator] 1a
89 @property 1a
90 def public_url(self) -> str: 1a
91 return S3_SERVICES[self.service].get_public_url(self.path)
94class ProductMediaFileRead(PublicFileReadBase): 1a
95 """File to be used as a product media file."""
97 service: Literal[FileServiceTypes.product_media] 1a
100class OrganizationAvatarFileRead(PublicFileReadBase): 1a
101 """File to be used as an organization avatar."""
103 service: Literal[FileServiceTypes.organization_avatar] 1a
106FileRead = Annotated[ 1a
107 DownloadableFileRead | ProductMediaFileRead | OrganizationAvatarFileRead,
108 Discriminator("service"),
109 MergeJSONSchema({"title": "FileRead"}),
110 ClassName("FileRead"),
111]
113FileReadAdapter: TypeAdapter[FileRead] = TypeAdapter[FileRead](FileRead) 1a
116class FileUpload(S3FileUpload): 1a
117 version: str | None 1a
118 is_uploaded: bool = False 1a
119 service: FileServiceTypes 1a
122class FileUploadCompleted(S3FileUploadCompleted): ... 1a
125class FileDownload(S3FileDownload): 1a
126 version: str | None 1a
127 is_uploaded: bool 1a
128 service: FileServiceTypes 1a
130 @classmethod 1a
131 def from_presigned(cls, file: File, url: str, expires_at: datetime) -> Self: 1a
132 file_dict: dict[str, Any] = dict(
133 id=file.id,
134 organization_id=file.organization_id,
135 name=file.name,
136 path=file.path,
137 mime_type=file.mime_type,
138 size=file.size,
139 version=file.version,
140 service=file.service,
141 checksum_etag=file.checksum_etag,
142 last_modified_at=file.last_modified_at,
143 storage_version=file.storage_version,
144 is_uploaded=file.is_uploaded,
145 created_at=file.created_at,
146 )
147 if file.checksum_sha256_base64 and file.checksum_sha256_hex:
148 file_dict.update(
149 checksum_sha256_base64=file.checksum_sha256_base64,
150 checksum_sha256_hex=file.checksum_sha256_hex,
151 )
153 return cls(
154 **file_dict,
155 download=S3DownloadURL(
156 url=url,
157 expires_at=expires_at,
158 ),
159 )
162class FileUpdate(Schema): 1a
163 id: UUID4 1a
164 version: str | None 1a
165 checksum_etag: str 1a
166 last_modified_at: datetime 1a
167 storage_version: str | None 1a
168 checksum_sha256_base64: str | None 1a
169 checksum_sha256_hex: str | None 1a
172class FilePatch(Schema): 1a
173 name: str | None = None 1a
174 version: str | None = None 1a