(self, request, identifier, *args, **kwargs)
| 111 | |
| 112 | # identifier is in a path parameter |
| 113 | def get(self, request, identifier, *args, **kwargs): # type: ignore[no-untyped-def] |
| 114 | # if we have identifier fetch, or create if does not exist |
| 115 | identity, is_new_identity = Identity.objects.get_or_create_for_sdk( |
| 116 | identifier=identifier, |
| 117 | environment=request.environment, |
| 118 | ) |
| 119 | |
| 120 | # New identities may take a while to replicate — otherwise use a replica |
| 121 | if not is_new_identity and is_database_replica_setup(): |
| 122 | identity = ( |
| 123 | using_database_replica(Identity.objects) |
| 124 | .with_context() |
| 125 | .get(id=identity.id) |
| 126 | ) |
| 127 | |
| 128 | traits_data = identity.get_all_user_traits() # type: ignore[no-untyped-call] |
| 129 | |
| 130 | # We need object type to pass into our IdentitySerializerTraitFlags |
| 131 | IdentityFlagsWithTraitsAndSegments = namedtuple( # type: ignore[name-match] |
| 132 | "IdentityTraitFlagsSegments", ("flags", "traits", "segments") |
| 133 | ) |
| 134 | identity_flags_traits_segments = IdentityFlagsWithTraitsAndSegments( |
| 135 | flags=identity.get_all_feature_states(), |
| 136 | traits=traits_data, |
| 137 | segments=identity.get_segments(), |
| 138 | ) |
| 139 | |
| 140 | serializer = IdentitySerializerWithTraitsAndSegments( |
| 141 | identity_flags_traits_segments |
| 142 | ) |
| 143 | |
| 144 | return Response(serializer.data, status=status.HTTP_200_OK) |
| 145 | |
| 146 | |
| 147 | @extend_schema(tags=["sdk"]) |
nothing calls this directly
no test coverage detected