(
self, kwargs: Mapping[str, Any], coll_name: str, ask_db: bool
)
| 304 | ) |
| 305 | |
| 306 | def _get_encrypted_fields( |
| 307 | self, kwargs: Mapping[str, Any], coll_name: str, ask_db: bool |
| 308 | ) -> Optional[Mapping[str, Any]]: |
| 309 | encrypted_fields = kwargs.get("encryptedFields") |
| 310 | if encrypted_fields: |
| 311 | return cast(Mapping[str, Any], deepcopy(encrypted_fields)) |
| 312 | if ( |
| 313 | self.client.options.auto_encryption_opts |
| 314 | and self.client.options.auto_encryption_opts._encrypted_fields_map |
| 315 | and self.client.options.auto_encryption_opts._encrypted_fields_map.get( |
| 316 | f"{self.name}.{coll_name}" |
| 317 | ) |
| 318 | ): |
| 319 | return cast( |
| 320 | Mapping[str, Any], |
| 321 | deepcopy( |
| 322 | self.client.options.auto_encryption_opts._encrypted_fields_map[ |
| 323 | f"{self.name}.{coll_name}" |
| 324 | ] |
| 325 | ), |
| 326 | ) |
| 327 | if ask_db and self.client.options.auto_encryption_opts: |
| 328 | options = self[coll_name].options() |
| 329 | if options.get("encryptedFields"): |
| 330 | return cast(Mapping[str, Any], deepcopy(options["encryptedFields"])) |
| 331 | return None |
| 332 | |
| 333 | # See PYTHON-3084. |
| 334 | __iter__ = None |
no test coverage detected