Coverage for functions \ flipdare \ generated \ model \ backend \ compliance_model.py: 100%
0 statements
« prev ^ index » next coverage.py v7.13.0, created at 2026-05-08 12:22 +1000
1#!/usr/bin/env python
2#
3# Copyright (c) 2026 Flipdare Pty Ltd. All rights reserved.
4#
5# This file is part of Flipdare's proprietary software and contains
6# confidential and copyrighted material. Unauthorised copying,
7# modification, distribution, or use of this file is strictly
8# prohibited without prior written permission from Flipdare Pty Ltd.
9#
10# This software includes third-party components licensed under MIT,
11# BSD, and Apache 2.0 licences. See THIRD_PARTY_NOTICES for details.
12#
13# NOTE: THIS FILE IS AUTO GENERATED. DO NOT EDIT.
14#
15# Generated by codegen_models.py
16#
17# Modify 'codegen_models.py'
18# and re-run the script above to update.
19#
20# pragma: no cover
21from __future__ import annotations
22from datetime import datetime
23from google.cloud.firestore_v1.transforms import Sentinel
24from flipdare.core.firestore_field import FirestoreField
25from flipdare.util.time_util import FirestoreTime
26from typing import Any, TypedDict, cast, Unpack
27from enum import StrEnum
28from pydantic import Field, ConfigDict, TypeAdapter
29from flipdare.firestore.core.app_base_model import AppBaseModel
30from flipdare.generated.shared.model.compliance_obj_type import ComplianceObjType
31from flipdare.generated.shared.model.compliance_obj_type import ComplianceObjType
32from flipdare.wrapper import (
33 ChatWrapper,
34 ChatCommentWrapper,
35 DareWrapper,
36 GroupWrapper,
37 ContentWrapper,
38 PledgeWrapper,
39 UserWrapper,
40 PersistedWrapper,
41)
44class ComplianceKeys(StrEnum):
45 ID = "id"
46 CREATED_AT = "created_at"
47 UPDATED_AT = "updated_at"
48 UID = "uid"
49 DELETED_BY_UID = "deleted_by_uid"
50 OBJ_ID = "obj_id"
51 OBJ_TYPE = "obj_type"
52 OBJ_MODEL = "obj_model"
53 VERSION = "version"
54 PROCESSED = "processed"
55 ERROR_COUNT = "error_count"
56 ANONYMIZED = "anonymized"
57 DELETED_FROM_SEARCH = "deleted_from_search"
60# !! IMPORTANT !!
61# !!
62# !! this should only be used in the database to query.
63# !!
64class ComplianceInternalKeys(StrEnum):
65 CREATED_AT = "created_at"
66 UPDATED_AT = "updated_at"
67 VERSION = "VERSION"
68 PROCESSED = "INT_P"
69 ERROR_COUNT = "INT_E"
70 ANONYMIZED = "INT_A"
71 DELETED_FROM_SEARCH = "INT_DFS"
74class ComplianceModel(AppBaseModel):
75 """Stores all deleted objects for compliance purposes."""
77 model_config = ConfigDict(populate_by_name=True, arbitrary_types_allowed=True)
79 id: str | None = Field(None, alias="id")
80 created_at: FirestoreField = Field(
81 default_factory=cast("Any", lambda: FirestoreTime.server_timestamp())
82 )
83 updated_at: FirestoreField = Field(
84 default_factory=cast("Any", lambda: FirestoreTime.server_timestamp())
85 )
86 uid: str
87 deleted_by_uid: str
88 obj_id: str
89 obj_type: ComplianceObjType
90 obj_model: dict[str, Any]
91 # Version (base internal field)
92 version: int = Field(default=1, alias="VERSION")
93 # Processed (base internal field)
94 processed: bool = Field(default=False, alias="INT_P")
95 # Error Count (base internal field)
96 error_count: int = Field(default=0, alias="INT_E")
97 # Anonymized (base internal field)
98 anonymized: bool = Field(default=False, alias="INT_A")
99 # Deleted from Search (base internal field)
100 deleted_from_search: bool = Field(default=False, alias="INT_DFS")
102 @classmethod
103 def validate_partial(cls, **data: Unpack[ComplianceDict]) -> dict[str, Any]:
104 """
105 Uses Unpack to give you autocomplete and static warnings
106 if you pass an invalid key or type in your code.
108 Returns a dict with Firestore field names (aliases) for use with batch.update().
109 """
110 result: dict[str, Any] = {}
111 for k, v in data.items():
112 if k in cls.__pydantic_fields__:
113 field_info = cls.__pydantic_fields__[k]
114 validated_value = cast(
115 "Any",
116 TypeAdapter(field_info.annotation).validate_python(v),
117 )
118 # Use alias if defined, otherwise use field name
119 output_key = field_info.alias or k
120 result[output_key] = validated_value
121 return result
123 # ---- Convenience factories -----------------------------------------
124 @classmethod
125 def from_wrapper(
126 cls,
127 wrapper: PersistedWrapper[Any],
128 deleted_by_uid: str | None = None,
129 ) -> ComplianceModel:
130 match wrapper:
131 case ChatWrapper():
132 return cls.from_chat(wrapper, deleted_by_uid)
133 case GroupWrapper():
134 return cls.from_group(wrapper, deleted_by_uid)
135 case DareWrapper():
136 return cls.from_dare(wrapper, deleted_by_uid)
137 case UserWrapper():
138 return cls.from_user(wrapper, deleted_by_uid)
139 case PledgeWrapper():
140 return cls.from_pledge(wrapper, deleted_by_uid)
141 case ContentWrapper():
142 return cls.from_content(wrapper, deleted_by_uid)
143 case ChatCommentWrapper():
144 return cls.from_chat_message(wrapper, deleted_by_uid)
145 case _:
146 msg = f"Unsupported wrapper type for compliance: {type(wrapper).__name__}"
147 raise ValueError(msg)
149 @classmethod
150 def from_chat(
151 cls,
152 chat: ChatWrapper,
153 deleted_by_uid: str | None = None,
154 ) -> ComplianceModel:
155 uid = chat.uid
156 return cls(
157 id=None,
158 obj_id=chat.doc_id,
159 uid=uid,
160 obj_model=chat.to_dict(),
161 obj_type=ComplianceObjType.CHAT,
162 deleted_by_uid=deleted_by_uid or uid,
163 )
165 @classmethod
166 def from_chat_message(
167 cls,
168 message: ChatCommentWrapper,
169 deleted_by_uid: str | None = None,
170 ) -> ComplianceModel:
171 uid = message.from_uid
172 return cls(
173 id=None,
174 obj_id=message.doc_id,
175 uid=uid,
176 obj_model=message.to_dict(),
177 obj_type=ComplianceObjType.CHAT_MESSAGE,
178 deleted_by_uid=deleted_by_uid or uid,
179 )
181 @classmethod
182 def from_group(
183 cls,
184 group: GroupWrapper,
185 deleted_by_uid: str | None = None,
186 ) -> ComplianceModel:
187 uid = group.uid
188 return cls(
189 id=None,
190 obj_id=group.doc_id,
191 uid=uid,
192 obj_model=group.to_dict(),
193 obj_type=ComplianceObjType.GROUP,
194 deleted_by_uid=deleted_by_uid or uid,
195 )
197 @classmethod
198 def from_dare(
199 cls,
200 dare: DareWrapper,
201 deleted_by_uid: str | None = None,
202 ) -> ComplianceModel:
203 uid = dare.from_uid
204 return cls(
205 id=None,
206 obj_id=dare.doc_id,
207 uid=uid,
208 obj_model=dare.to_dict(),
209 obj_type=ComplianceObjType.DARE,
210 deleted_by_uid=deleted_by_uid or uid,
211 )
213 @classmethod
214 def from_user(
215 cls,
216 user: UserWrapper,
217 deleted_by_uid: str | None = None,
218 ) -> ComplianceModel:
219 uid = user.doc_id
220 # Users should be anonymized instead of deleted for compliance purposes,
221 # so we set must_anonymize to True.
222 return cls(
223 id=None,
224 obj_id=uid,
225 uid=uid,
226 obj_model=user.to_dict(),
227 obj_type=ComplianceObjType.USER,
228 deleted_by_uid=deleted_by_uid or uid,
229 )
231 @classmethod
232 def from_pledge(
233 cls,
234 pledge: PledgeWrapper,
235 deleted_by_uid: str | None = None,
236 ) -> ComplianceModel:
237 uid = pledge.from_uid
238 return cls(
239 id=None,
240 obj_id=pledge.doc_id,
241 uid=uid,
242 obj_model=pledge.to_dict(),
243 obj_type=ComplianceObjType.PLEDGE,
244 deleted_by_uid=deleted_by_uid or uid,
245 )
247 @classmethod
248 def from_content(
249 cls,
250 content: ContentWrapper,
251 deleted_by_uid: str | None = None,
252 ) -> ComplianceModel:
253 uid = content.uid
254 return cls(
255 id=None,
256 obj_id=content.doc_id,
257 uid=uid,
258 obj_model=content.to_dict(),
259 obj_type=ComplianceObjType.CONTENT,
260 deleted_by_uid=deleted_by_uid or uid,
261 )
264COMPLIANCE_FIELD_NAMES: list[str] = list(ComplianceModel.model_fields.keys())
267class ComplianceDict(TypedDict, total=False):
268 id: str | None
269 created_at: Sentinel | datetime | str
270 updated_at: Sentinel | datetime | str
271 uid: str
272 deleted_by_uid: str
273 obj_id: str
274 obj_type: ComplianceObjType
275 obj_model: dict[str, Any]
276 VERSION: int | None
277 INT_P: bool | None
278 INT_E: int | None
279 INT_A: bool | None
280 INT_DFS: bool | None