Refactor get_or_create_person function in process_event (#1626)

* Refactor get_or_create_person function in process_event

* optimize get_or_create person

.github/workflows/ci-backend.yml

@@ -143,15 +143,6 @@ jobs:
         runs-on: ubuntu-latest
 
         services:
-            redis:
-                # Docker Hub image
-                image: redis
-                # Set health checks to wait until redis has started
-                options: >-
-                    --health-cmd "redis-cli ping"
-                    --health-interval 10s
-                    --health-timeout 5s
-                    --health-retries 5
             postgres:
                 image: postgres:12
                 env:

ee/clickhouse/process_event.py

@@ -16,7 +16,7 @@ from posthog.ee import check_ee_enabled
 from posthog.models.element import Element
 from posthog.models.person import Person
 from posthog.models.team import Team
-from posthog.tasks.process_event import check_and_create_person, handle_timestamp, store_names_and_properties
+from posthog.tasks.process_event import get_or_create_person, handle_timestamp, store_names_and_properties
 
 
 def _alias(previous_distinct_id: str, distinct_id: str, team_id: int, retry_if_failed: bool = True,) -> None:
@@ -124,8 +124,8 @@ def _capture_ee(
     )
 
     # # check/create persondistinctid
-    person = check_and_create_person(team_id=team.pk, distinct_id=distinct_id)
-    if person:
+    person, created = get_or_create_person(team_id=team.pk, distinct_id=distinct_id)
+    if created:
         create_person_with_distinct_id(person_id=person.pk, distinct_ids=[distinct_id], team_id=team.pk)
 
 

posthog/tasks/process_event.py

@@ -1,6 +1,6 @@
 import datetime
 from numbers import Number
-from typing import Dict, Optional, Union
+from typing import Dict, Optional, Tuple, Union
 
 import posthoganalytics
 from celery import shared_task
@@ -131,21 +131,36 @@ def _capture(
         **({"elements": elements_list} if elements_list else {})
     )
     store_names_and_properties(team=team, event=event, properties=properties)
-    check_and_create_person(team_id=team_id, distinct_id=distinct_id)
+    check_created_person(team_id=team_id, distinct_id=distinct_id)
 
 
-def check_and_create_person(team_id: int, distinct_id: str) -> Optional[Person]:
-    person: Optional[Person]
+def check_created_person(team_id: int, distinct_id: str) -> None:
+    if not Person.objects.distinct_ids_exist(team_id=team_id, distinct_ids=[str(distinct_id)]):
+        # Catch race condition where in between getting and creating,
+        # another request already created this user
+        try:
+            Person.objects.create(team_id=team_id, distinct_ids=[str(distinct_id)])
+        except IntegrityError:
+            return
+    return
+
+
+def get_or_create_person(team_id: int, distinct_id: str) -> Tuple[Person, bool]:
+    person: Person
+    created = False
+
     if not Person.objects.distinct_ids_exist(team_id=team_id, distinct_ids=[str(distinct_id)]):
-        # Catch race condition where in between getting and creating, another request already created this user.
         try:
             person = Person.objects.create(team_id=team_id, distinct_ids=[str(distinct_id)])
+            created = True
         except IntegrityError:
-            return None
-        return person
+            person = Person.objects.get(team_id=team_id, persondistinctid__distinct_id=str(distinct_id))
+            created = False
     else:
         person = Person.objects.get(team_id=team_id, persondistinctid__distinct_id=str(distinct_id))
-        return person
+        created = False
+
+    return person, created
 
 
 def _update_person_properties(team_id: int, distinct_id: str, properties: Dict) -> None: