Find unique indexes in the document schema and return them.
(cls, namespace="")
| 970 | |
| 971 | @classmethod |
| 972 | def _unique_with_indexes(cls, namespace=""): |
| 973 | """Find unique indexes in the document schema and return them.""" |
| 974 | unique_indexes = [] |
| 975 | for field_name, field in cls._fields.items(): |
| 976 | sparse = field.sparse |
| 977 | |
| 978 | # Generate a list of indexes needed by uniqueness constraints |
| 979 | if field.unique: |
| 980 | unique_fields = [field.db_field] |
| 981 | |
| 982 | # Add any unique_with fields to the back of the index spec |
| 983 | if field.unique_with: |
| 984 | if isinstance(field.unique_with, str): |
| 985 | field.unique_with = [field.unique_with] |
| 986 | |
| 987 | # Convert unique_with field names to real field names |
| 988 | unique_with = [] |
| 989 | for other_name in field.unique_with: |
| 990 | parts = other_name.split(".") |
| 991 | |
| 992 | # Lookup real name |
| 993 | parts = cls._lookup_field(parts) |
| 994 | name_parts = [part.db_field for part in parts] |
| 995 | unique_with.append(".".join(name_parts)) |
| 996 | |
| 997 | # Unique field should be required |
| 998 | parts[-1].required = True |
| 999 | sparse = not sparse and parts[-1].name not in cls.__dict__ |
| 1000 | |
| 1001 | unique_fields += unique_with |
| 1002 | |
| 1003 | # Add the new index to the list |
| 1004 | fields = [(f"{namespace}{f}", pymongo.ASCENDING) for f in unique_fields] |
| 1005 | index = {"fields": fields, "unique": True, "sparse": sparse} |
| 1006 | unique_indexes.append(index) |
| 1007 | |
| 1008 | if field.__class__.__name__ in { |
| 1009 | "EmbeddedDocumentListField", |
| 1010 | "ListField", |
| 1011 | "SortedListField", |
| 1012 | }: |
| 1013 | field = field.field |
| 1014 | |
| 1015 | # Grab any embedded document field unique indexes |
| 1016 | if ( |
| 1017 | field.__class__.__name__ == "EmbeddedDocumentField" |
| 1018 | and field.document_type != cls |
| 1019 | ): |
| 1020 | field_namespace = "%s." % field_name |
| 1021 | doc_cls = field.document_type |
| 1022 | unique_indexes += doc_cls._unique_with_indexes(field_namespace) |
| 1023 | |
| 1024 | return unique_indexes |
| 1025 | |
| 1026 | @classmethod |
| 1027 | def _geo_indices(cls, inspected=None, parent_field=None): |
no test coverage detected