We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Fix typing for `sentry.tasks.commits`
Many files related to post-process are skipping type checks.

* Remove the file from [sentry/pyproject.toml](https://github.com/getsentry/sentry/blob/b7810a9dca4b57afd2858903a6a9ec7ab50cdead/pyproject.toml)
* Run `mypy sentry.tasks.commits` and fix the typing errors

From https://github.com/getsentry/sentry/issues/55193

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `src/sentry/tasks/commits.py`
Content:
```
1 from __future__ import annotations
2 
3 import logging
4 
5 import sentry_sdk
6 from django.urls import reverse
7 from sentry_sdk import set_tag
8 
9 from sentry.constants import ObjectStatus
10 from sentry.exceptions import InvalidIdentity, PluginError
11 from sentry.models.deploy import Deploy
12 from sentry.models.latestreporeleaseenvironment import LatestRepoReleaseEnvironment
13 from sentry.models.organization import Organization
14 from sentry.models.release import Release, ReleaseCommitError
15 from sentry.models.releaseheadcommit import ReleaseHeadCommit
16 from sentry.models.repository import Repository
17 from sentry.models.user import User
18 from sentry.plugins.base import bindings
19 from sentry.services.hybrid_cloud.user import RpcUser
20 from sentry.services.hybrid_cloud.user.service import user_service
21 from sentry.shared_integrations.exceptions import IntegrationError
22 from sentry.silo import SiloMode
23 from sentry.tasks.base import instrumented_task, retry
24 from sentry.utils.email import MessageBuilder
25 from sentry.utils.http import absolute_uri
26 
27 logger = logging.getLogger(__name__)
28 
29 
30 def generate_invalid_identity_email(identity, commit_failure=False):
31     new_context = {
32         "identity": identity,
33         "auth_url": absolute_uri(reverse("socialauth_associate", args=[identity.provider])),
34         "commit_failure": commit_failure,
35     }
36 
37     return MessageBuilder(
38         subject="Unable to Fetch Commits" if commit_failure else "Action Required",
39         context=new_context,
40         template="sentry/emails/identity-invalid.txt",
41         html_template="sentry/emails/identity-invalid.html",
42     )
43 
44 
45 def generate_fetch_commits_error_email(release, repo, error_message):
46     new_context = {"release": release, "error_message": error_message, "repo": repo}
47 
48     return MessageBuilder(
49         subject="Unable to Fetch Commits",
50         context=new_context,
51         template="sentry/emails/unable-to-fetch-commits.txt",
52         html_template="sentry/emails/unable-to-fetch-commits.html",
53     )
54 
55 
56 # we're future proofing this function a bit so it could be used with other code
57 
58 
59 def handle_invalid_identity(identity, commit_failure=False):
60     # email the user
61     msg = generate_invalid_identity_email(identity, commit_failure)
62     msg.send_async(to=[identity.user.email])
63 
64     # now remove the identity, as its invalid
65     identity.delete()
66 
67 
68 @instrumented_task(
69     name="sentry.tasks.commits.fetch_commits",
70     queue="commits",
71     default_retry_delay=60 * 5,
72     max_retries=5,
73     silo_mode=SiloMode.REGION,
74 )
75 @retry(exclude=(Release.DoesNotExist, User.DoesNotExist))
76 def fetch_commits(release_id: int, user_id: int, refs, prev_release_id=None, **kwargs):
77     # TODO(dcramer): this function could use some cleanup/refactoring as it's a bit unwieldy
78     commit_list = []
79 
80     release = Release.objects.get(id=release_id)
81     set_tag("organization.slug", release.organization.slug)
82     # TODO: Need a better way to error handle no user_id. We need the SDK to be able to call this without user context
83     # to autoassociate commits to releases
84     user = user_service.get_user(user_id) if user_id is not None else None
85     # user = User.objects.get(id=user_id) if user_id is not None else None
86     prev_release = None
87     if prev_release_id is not None:
88         try:
89             prev_release = Release.objects.get(id=prev_release_id)
90         except Release.DoesNotExist:
91             pass
92 
93     for ref in refs:
94         repo = (
95             Repository.objects.filter(
96                 organization_id=release.organization_id,
97                 name=ref["repository"],
98                 status=ObjectStatus.ACTIVE,
99             )
100             .order_by("-pk")
101             .first()
102         )
103         if not repo:
104             logger.info(
105                 "repository.missing",
106                 extra={
107                     "organization_id": release.organization_id,
108                     "user_id": user_id,
109                     "repository": ref["repository"],
110                 },
111             )
112             continue
113 
114         binding_key = (
115             "integration-repository.provider"
116             if is_integration_provider(repo.provider)
117             else "repository.provider"
118         )
119         try:
120             provider_cls = bindings.get(binding_key).get(repo.provider)
121         except KeyError:
122             continue
123 
124         # if previous commit isn't provided, try to get from
125         # previous release otherwise, try to get
126         # recent commits from provider api
127         start_sha = None
128         if ref.get("previousCommit"):
129             start_sha = ref["previousCommit"]
130         elif prev_release:
131             try:
132                 start_sha = ReleaseHeadCommit.objects.filter(
133                     organization_id=release.organization_id,
134                     release=prev_release,
135                     repository_id=repo.id,
136                 ).values_list("commit__key", flat=True)[0]
137             except IndexError:
138                 pass
139 
140         end_sha = ref["commit"]
141         provider = provider_cls(id=repo.provider)
142         try:
143             if is_integration_provider(provider.id):
144                 repo_commits = provider.compare_commits(repo, start_sha, end_sha)
145             else:
146                 repo_commits = provider.compare_commits(repo, start_sha, end_sha, actor=user)
147         except NotImplementedError:
148             pass
149         except Exception as e:
150             logger.info(
151                 "fetch_commits.error",
152                 extra={
153                     "organization_id": repo.organization_id,
154                     "user_id": user_id,
155                     "repository": repo.name,
156                     "provider": provider.id,
157                     "error": str(e),
158                     "end_sha": end_sha,
159                     "start_sha": start_sha,
160                 },
161             )
162             span = sentry_sdk.Hub.current.scope.span
163             span.set_status("unknown_error")
164             logger.exception(e)
165             if isinstance(e, InvalidIdentity) and getattr(e, "identity", None):
166                 handle_invalid_identity(identity=e.identity, commit_failure=True)
167             elif isinstance(e, (PluginError, InvalidIdentity, IntegrationError)):
168                 msg = generate_fetch_commits_error_email(release, repo, str(e))
169                 emails = get_emails_for_user_or_org(user, release.organization_id)
170                 msg.send_async(to=emails)
171             else:
172                 msg = generate_fetch_commits_error_email(
173                     release, repo, "An internal system error occurred."
174                 )
175                 emails = get_emails_for_user_or_org(user, release.organization_id)
176                 msg.send_async(to=emails)
177         else:
178             logger.info(
179                 "fetch_commits.complete",
180                 extra={
181                     "organization_id": repo.organization_id,
182                     "user_id": user_id,
183                     "repository": repo.name,
184                     "end_sha": end_sha,
185                     "start_sha": start_sha,
186                     "num_commits": len(repo_commits or []),
187                 },
188             )
189             commit_list.extend(repo_commits)
190 
191     if commit_list:
192         try:
193             release.set_commits(commit_list)
194         except ReleaseCommitError:
195             # Another task or webworker is currently setting commits on this
196             # release. Return early as that task will do the remaining work.
197             logger.info(
198                 "fetch_commits.duplicate",
199                 extra={
200                     "release_id": release.id,
201                     "organization_id": release.organization_id,
202                     "user_id": user_id,
203                 },
204             )
205             return
206 
207         deploys = Deploy.objects.filter(
208             organization_id=release.organization_id, release=release, notified=False
209         ).values_list("id", "environment_id", "date_finished")
210 
211         # XXX(dcramer): i don't know why this would have multiple environments, but for
212         # our sanity lets assume it can
213         pending_notifications = []
214         last_deploy_per_environment = {}
215         for deploy_id, environment_id, date_finished in deploys:
216             last_deploy_per_environment[environment_id] = (deploy_id, date_finished)
217             pending_notifications.append(deploy_id)
218 
219         repo_queryset = ReleaseHeadCommit.objects.filter(
220             organization_id=release.organization_id, release=release
221         ).values_list("repository_id", "commit")
222 
223         # for each repo, update (or create if this is the first one) our records
224         # of the latest commit-associated release in each env
225         # use deploys as a proxy for ReleaseEnvironment, because they contain
226         # a timestamp in addition to release and env data
227         for repository_id, commit_id in repo_queryset:
228             for environment_id, (deploy_id, date_finished) in last_deploy_per_environment.items():
229                 # we need to mark LatestRepoReleaseEnvironment, but only if there's not a
230                 # deploy in the given environment which has completed *after*
231                 # this deploy (given we might process commits out of order)
232                 if not Deploy.objects.filter(
233                     id__in=LatestRepoReleaseEnvironment.objects.filter(
234                         repository_id=repository_id, environment_id=environment_id
235                     ).values("deploy_id"),
236                     date_finished__gt=date_finished,
237                 ).exists():
238                     LatestRepoReleaseEnvironment.objects.create_or_update(
239                         repository_id=repository_id,
240                         environment_id=environment_id,
241                         values={
242                             "release_id": release.id,
243                             "deploy_id": deploy_id,
244                             "commit_id": commit_id,
245                         },
246                     )
247 
248         for deploy_id in pending_notifications:
249             Deploy.notify_if_ready(deploy_id, fetch_complete=True)
250 
251 
252 def is_integration_provider(provider):
253     return provider and provider.startswith("integrations:")
254 
255 
256 def get_emails_for_user_or_org(user: RpcUser | None, orgId: int):
257     emails = []
258     if not user:
259         return []
260     if user.is_sentry_app:
261         organization = Organization.objects.get(id=orgId)
262         members = organization.get_members_with_org_roles(roles=["owner"])
263         user_ids = [m.user_id for m in members if m.user_id]
264         emails = {u.email for u in user_service.get_many(filter={"user_ids": user_ids}) if u.email}
265         emails = list(emails)
266     else:
267         emails = [user.email]
268 
269     return emails
270
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/src/sentry/tasks/commits.py b/src/sentry/tasks/commits.py
--- a/src/sentry/tasks/commits.py
+++ b/src/sentry/tasks/commits.py
@@ -159,7 +159,9 @@
                     "start_sha": start_sha,
                 },
             )
-            span = sentry_sdk.Hub.current.scope.span
+            span = sentry_sdk.get_current_span()
+            if span is None:
+                raise TypeError("No span is currently active right now")
             span.set_status("unknown_error")
             logger.exception(e)
             if isinstance(e, InvalidIdentity) and getattr(e, "identity", None):
@@ -254,15 +256,16 @@
 
 
 def get_emails_for_user_or_org(user: RpcUser | None, orgId: int):
-    emails = []
+    emails: list[str] = []
     if not user:
         return []
     if user.is_sentry_app:
         organization = Organization.objects.get(id=orgId)
         members = organization.get_members_with_org_roles(roles=["owner"])
         user_ids = [m.user_id for m in members if m.user_id]
-        emails = {u.email for u in user_service.get_many(filter={"user_ids": user_ids}) if u.email}
-        emails = list(emails)
+        emails = list(
+            {u.email for u in user_service.get_many(filter={"user_ids": user_ids}) if u.email}
+        )
     else:
         emails = [user.email]

{"golden_diff": "diff --git a/src/sentry/tasks/commits.py b/src/sentry/tasks/commits.py\n--- a/src/sentry/tasks/commits.py\n+++ b/src/sentry/tasks/commits.py\n@@ -159,7 +159,9 @@\n                     \"start_sha\": start_sha,\n                 },\n             )\n-            span = sentry_sdk.Hub.current.scope.span\n+            span = sentry_sdk.get_current_span()\n+            if span is None:\n+                raise TypeError(\"No span is currently active right now\")\n             span.set_status(\"unknown_error\")\n             logger.exception(e)\n             if isinstance(e, InvalidIdentity) and getattr(e, \"identity\", None):\n@@ -254,15 +256,16 @@\n \n \n def get_emails_for_user_or_org(user: RpcUser | None, orgId: int):\n-    emails = []\n+    emails: list[str] = []\n     if not user:\n         return []\n     if user.is_sentry_app:\n         organization = Organization.objects.get(id=orgId)\n         members = organization.get_members_with_org_roles(roles=[\"owner\"])\n         user_ids = [m.user_id for m in members if m.user_id]\n-        emails = {u.email for u in user_service.get_many(filter={\"user_ids\": user_ids}) if u.email}\n-        emails = list(emails)\n+        emails = list(\n+            {u.email for u in user_service.get_many(filter={\"user_ids\": user_ids}) if u.email}\n+        )\n     else:\n         emails = [user.email]\n", "issue": "Fix typing for `sentry.tasks.commits`\nMany files related to post-process are skipping type checks.\n\n* Remove the file from [sentry/pyproject.toml](https://github.com/getsentry/sentry/blob/b7810a9dca4b57afd2858903a6a9ec7ab50cdead/pyproject.toml)\n* Run `mypy sentry.tasks.commits` and fix the typing errors\n\nFrom https://github.com/getsentry/sentry/issues/55193\n", "before_files": [{"content": "from __future__ import annotations\n\nimport logging\n\nimport sentry_sdk\nfrom django.urls import reverse\nfrom sentry_sdk import set_tag\n\nfrom sentry.constants import ObjectStatus\nfrom sentry.exceptions import InvalidIdentity, PluginError\nfrom sentry.models.deploy import Deploy\nfrom sentry.models.latestreporeleaseenvironment import LatestRepoReleaseEnvironment\nfrom sentry.models.organization import Organization\nfrom sentry.models.release import Release, ReleaseCommitError\nfrom sentry.models.releaseheadcommit import ReleaseHeadCommit\nfrom sentry.models.repository import Repository\nfrom sentry.models.user import User\nfrom sentry.plugins.base import bindings\nfrom sentry.services.hybrid_cloud.user import RpcUser\nfrom sentry.services.hybrid_cloud.user.service import user_service\nfrom sentry.shared_integrations.exceptions import IntegrationError\nfrom sentry.silo import SiloMode\nfrom sentry.tasks.base import instrumented_task, retry\nfrom sentry.utils.email import MessageBuilder\nfrom sentry.utils.http import absolute_uri\n\nlogger = logging.getLogger(__name__)\n\n\ndef generate_invalid_identity_email(identity, commit_failure=False):\n    new_context = {\n        \"identity\": identity,\n        \"auth_url\": absolute_uri(reverse(\"socialauth_associate\", args=[identity.provider])),\n        \"commit_failure\": commit_failure,\n    }\n\n    return MessageBuilder(\n        subject=\"Unable to Fetch Commits\" if commit_failure else \"Action Required\",\n        context=new_context,\n        template=\"sentry/emails/identity-invalid.txt\",\n        html_template=\"sentry/emails/identity-invalid.html\",\n    )\n\n\ndef generate_fetch_commits_error_email(release, repo, error_message):\n    new_context = {\"release\": release, \"error_message\": error_message, \"repo\": repo}\n\n    return MessageBuilder(\n        subject=\"Unable to Fetch Commits\",\n        context=new_context,\n        template=\"sentry/emails/unable-to-fetch-commits.txt\",\n        html_template=\"sentry/emails/unable-to-fetch-commits.html\",\n    )\n\n\n# we're future proofing this function a bit so it could be used with other code\n\n\ndef handle_invalid_identity(identity, commit_failure=False):\n    # email the user\n    msg = generate_invalid_identity_email(identity, commit_failure)\n    msg.send_async(to=[identity.user.email])\n\n    # now remove the identity, as its invalid\n    identity.delete()\n\n\n@instrumented_task(\n    name=\"sentry.tasks.commits.fetch_commits\",\n    queue=\"commits\",\n    default_retry_delay=60 * 5,\n    max_retries=5,\n    silo_mode=SiloMode.REGION,\n)\n@retry(exclude=(Release.DoesNotExist, User.DoesNotExist))\ndef fetch_commits(release_id: int, user_id: int, refs, prev_release_id=None, **kwargs):\n    # TODO(dcramer): this function could use some cleanup/refactoring as it's a bit unwieldy\n    commit_list = []\n\n    release = Release.objects.get(id=release_id)\n    set_tag(\"organization.slug\", release.organization.slug)\n    # TODO: Need a better way to error handle no user_id. We need the SDK to be able to call this without user context\n    # to autoassociate commits to releases\n    user = user_service.get_user(user_id) if user_id is not None else None\n    # user = User.objects.get(id=user_id) if user_id is not None else None\n    prev_release = None\n    if prev_release_id is not None:\n        try:\n            prev_release = Release.objects.get(id=prev_release_id)\n        except Release.DoesNotExist:\n            pass\n\n    for ref in refs:\n        repo = (\n            Repository.objects.filter(\n                organization_id=release.organization_id,\n                name=ref[\"repository\"],\n                status=ObjectStatus.ACTIVE,\n            )\n            .order_by(\"-pk\")\n            .first()\n        )\n        if not repo:\n            logger.info(\n                \"repository.missing\",\n                extra={\n                    \"organization_id\": release.organization_id,\n                    \"user_id\": user_id,\n                    \"repository\": ref[\"repository\"],\n                },\n            )\n            continue\n\n        binding_key = (\n            \"integration-repository.provider\"\n            if is_integration_provider(repo.provider)\n            else \"repository.provider\"\n        )\n        try:\n            provider_cls = bindings.get(binding_key).get(repo.provider)\n        except KeyError:\n            continue\n\n        # if previous commit isn't provided, try to get from\n        # previous release otherwise, try to get\n        # recent commits from provider api\n        start_sha = None\n        if ref.get(\"previousCommit\"):\n            start_sha = ref[\"previousCommit\"]\n        elif prev_release:\n            try:\n                start_sha = ReleaseHeadCommit.objects.filter(\n                    organization_id=release.organization_id,\n                    release=prev_release,\n                    repository_id=repo.id,\n                ).values_list(\"commit__key\", flat=True)[0]\n            except IndexError:\n                pass\n\n        end_sha = ref[\"commit\"]\n        provider = provider_cls(id=repo.provider)\n        try:\n            if is_integration_provider(provider.id):\n                repo_commits = provider.compare_commits(repo, start_sha, end_sha)\n            else:\n                repo_commits = provider.compare_commits(repo, start_sha, end_sha, actor=user)\n        except NotImplementedError:\n            pass\n        except Exception as e:\n            logger.info(\n                \"fetch_commits.error\",\n                extra={\n                    \"organization_id\": repo.organization_id,\n                    \"user_id\": user_id,\n                    \"repository\": repo.name,\n                    \"provider\": provider.id,\n                    \"error\": str(e),\n                    \"end_sha\": end_sha,\n                    \"start_sha\": start_sha,\n                },\n            )\n            span = sentry_sdk.Hub.current.scope.span\n            span.set_status(\"unknown_error\")\n            logger.exception(e)\n            if isinstance(e, InvalidIdentity) and getattr(e, \"identity\", None):\n                handle_invalid_identity(identity=e.identity, commit_failure=True)\n            elif isinstance(e, (PluginError, InvalidIdentity, IntegrationError)):\n                msg = generate_fetch_commits_error_email(release, repo, str(e))\n                emails = get_emails_for_user_or_org(user, release.organization_id)\n                msg.send_async(to=emails)\n            else:\n                msg = generate_fetch_commits_error_email(\n                    release, repo, \"An internal system error occurred.\"\n                )\n                emails = get_emails_for_user_or_org(user, release.organization_id)\n                msg.send_async(to=emails)\n        else:\n            logger.info(\n                \"fetch_commits.complete\",\n                extra={\n                    \"organization_id\": repo.organization_id,\n                    \"user_id\": user_id,\n                    \"repository\": repo.name,\n                    \"end_sha\": end_sha,\n                    \"start_sha\": start_sha,\n                    \"num_commits\": len(repo_commits or []),\n                },\n            )\n            commit_list.extend(repo_commits)\n\n    if commit_list:\n        try:\n            release.set_commits(commit_list)\n        except ReleaseCommitError:\n            # Another task or webworker is currently setting commits on this\n            # release. Return early as that task will do the remaining work.\n            logger.info(\n                \"fetch_commits.duplicate\",\n                extra={\n                    \"release_id\": release.id,\n                    \"organization_id\": release.organization_id,\n                    \"user_id\": user_id,\n                },\n            )\n            return\n\n        deploys = Deploy.objects.filter(\n            organization_id=release.organization_id, release=release, notified=False\n        ).values_list(\"id\", \"environment_id\", \"date_finished\")\n\n        # XXX(dcramer): i don't know why this would have multiple environments, but for\n        # our sanity lets assume it can\n        pending_notifications = []\n        last_deploy_per_environment = {}\n        for deploy_id, environment_id, date_finished in deploys:\n            last_deploy_per_environment[environment_id] = (deploy_id, date_finished)\n            pending_notifications.append(deploy_id)\n\n        repo_queryset = ReleaseHeadCommit.objects.filter(\n            organization_id=release.organization_id, release=release\n        ).values_list(\"repository_id\", \"commit\")\n\n        # for each repo, update (or create if this is the first one) our records\n        # of the latest commit-associated release in each env\n        # use deploys as a proxy for ReleaseEnvironment, because they contain\n        # a timestamp in addition to release and env data\n        for repository_id, commit_id in repo_queryset:\n            for environment_id, (deploy_id, date_finished) in last_deploy_per_environment.items():\n                # we need to mark LatestRepoReleaseEnvironment, but only if there's not a\n                # deploy in the given environment which has completed *after*\n                # this deploy (given we might process commits out of order)\n                if not Deploy.objects.filter(\n                    id__in=LatestRepoReleaseEnvironment.objects.filter(\n                        repository_id=repository_id, environment_id=environment_id\n                    ).values(\"deploy_id\"),\n                    date_finished__gt=date_finished,\n                ).exists():\n                    LatestRepoReleaseEnvironment.objects.create_or_update(\n                        repository_id=repository_id,\n                        environment_id=environment_id,\n                        values={\n                            \"release_id\": release.id,\n                            \"deploy_id\": deploy_id,\n                            \"commit_id\": commit_id,\n                        },\n                    )\n\n        for deploy_id in pending_notifications:\n            Deploy.notify_if_ready(deploy_id, fetch_complete=True)\n\n\ndef is_integration_provider(provider):\n    return provider and provider.startswith(\"integrations:\")\n\n\ndef get_emails_for_user_or_org(user: RpcUser | None, orgId: int):\n    emails = []\n    if not user:\n        return []\n    if user.is_sentry_app:\n        organization = Organization.objects.get(id=orgId)\n        members = organization.get_members_with_org_roles(roles=[\"owner\"])\n        user_ids = [m.user_id for m in members if m.user_id]\n        emails = {u.email for u in user_service.get_many(filter={\"user_ids\": user_ids}) if u.email}\n        emails = list(emails)\n    else:\n        emails = [user.email]\n\n    return emails\n", "path": "src/sentry/tasks/commits.py"}], "after_files": [{"content": "from __future__ import annotations\n\nimport logging\n\nimport sentry_sdk\nfrom django.urls import reverse\nfrom sentry_sdk import set_tag\n\nfrom sentry.constants import ObjectStatus\nfrom sentry.exceptions import InvalidIdentity, PluginError\nfrom sentry.models.deploy import Deploy\nfrom sentry.models.latestreporeleaseenvironment import LatestRepoReleaseEnvironment\nfrom sentry.models.organization import Organization\nfrom sentry.models.release import Release, ReleaseCommitError\nfrom sentry.models.releaseheadcommit import ReleaseHeadCommit\nfrom sentry.models.repository import Repository\nfrom sentry.models.user import User\nfrom sentry.plugins.base import bindings\nfrom sentry.services.hybrid_cloud.user import RpcUser\nfrom sentry.services.hybrid_cloud.user.service import user_service\nfrom sentry.shared_integrations.exceptions import IntegrationError\nfrom sentry.silo import SiloMode\nfrom sentry.tasks.base import instrumented_task, retry\nfrom sentry.utils.email import MessageBuilder\nfrom sentry.utils.http import absolute_uri\n\nlogger = logging.getLogger(__name__)\n\n\ndef generate_invalid_identity_email(identity, commit_failure=False):\n    new_context = {\n        \"identity\": identity,\n        \"auth_url\": absolute_uri(reverse(\"socialauth_associate\", args=[identity.provider])),\n        \"commit_failure\": commit_failure,\n    }\n\n    return MessageBuilder(\n        subject=\"Unable to Fetch Commits\" if commit_failure else \"Action Required\",\n        context=new_context,\n        template=\"sentry/emails/identity-invalid.txt\",\n        html_template=\"sentry/emails/identity-invalid.html\",\n    )\n\n\ndef generate_fetch_commits_error_email(release, repo, error_message):\n    new_context = {\"release\": release, \"error_message\": error_message, \"repo\": repo}\n\n    return MessageBuilder(\n        subject=\"Unable to Fetch Commits\",\n        context=new_context,\n        template=\"sentry/emails/unable-to-fetch-commits.txt\",\n        html_template=\"sentry/emails/unable-to-fetch-commits.html\",\n    )\n\n\n# we're future proofing this function a bit so it could be used with other code\n\n\ndef handle_invalid_identity(identity, commit_failure=False):\n    # email the user\n    msg = generate_invalid_identity_email(identity, commit_failure)\n    msg.send_async(to=[identity.user.email])\n\n    # now remove the identity, as its invalid\n    identity.delete()\n\n\n@instrumented_task(\n    name=\"sentry.tasks.commits.fetch_commits\",\n    queue=\"commits\",\n    default_retry_delay=60 * 5,\n    max_retries=5,\n    silo_mode=SiloMode.REGION,\n)\n@retry(exclude=(Release.DoesNotExist, User.DoesNotExist))\ndef fetch_commits(release_id: int, user_id: int, refs, prev_release_id=None, **kwargs):\n    # TODO(dcramer): this function could use some cleanup/refactoring as it's a bit unwieldy\n    commit_list = []\n\n    release = Release.objects.get(id=release_id)\n    set_tag(\"organization.slug\", release.organization.slug)\n    # TODO: Need a better way to error handle no user_id. We need the SDK to be able to call this without user context\n    # to autoassociate commits to releases\n    user = user_service.get_user(user_id) if user_id is not None else None\n    # user = User.objects.get(id=user_id) if user_id is not None else None\n    prev_release = None\n    if prev_release_id is not None:\n        try:\n            prev_release = Release.objects.get(id=prev_release_id)\n        except Release.DoesNotExist:\n            pass\n\n    for ref in refs:\n        repo = (\n            Repository.objects.filter(\n                organization_id=release.organization_id,\n                name=ref[\"repository\"],\n                status=ObjectStatus.ACTIVE,\n            )\n            .order_by(\"-pk\")\n            .first()\n        )\n        if not repo:\n            logger.info(\n                \"repository.missing\",\n                extra={\n                    \"organization_id\": release.organization_id,\n                    \"user_id\": user_id,\n                    \"repository\": ref[\"repository\"],\n                },\n            )\n            continue\n\n        binding_key = (\n            \"integration-repository.provider\"\n            if is_integration_provider(repo.provider)\n            else \"repository.provider\"\n        )\n        try:\n            provider_cls = bindings.get(binding_key).get(repo.provider)\n        except KeyError:\n            continue\n\n        # if previous commit isn't provided, try to get from\n        # previous release otherwise, try to get\n        # recent commits from provider api\n        start_sha = None\n        if ref.get(\"previousCommit\"):\n            start_sha = ref[\"previousCommit\"]\n        elif prev_release:\n            try:\n                start_sha = ReleaseHeadCommit.objects.filter(\n                    organization_id=release.organization_id,\n                    release=prev_release,\n                    repository_id=repo.id,\n                ).values_list(\"commit__key\", flat=True)[0]\n            except IndexError:\n                pass\n\n        end_sha = ref[\"commit\"]\n        provider = provider_cls(id=repo.provider)\n        try:\n            if is_integration_provider(provider.id):\n                repo_commits = provider.compare_commits(repo, start_sha, end_sha)\n            else:\n                repo_commits = provider.compare_commits(repo, start_sha, end_sha, actor=user)\n        except NotImplementedError:\n            pass\n        except Exception as e:\n            logger.info(\n                \"fetch_commits.error\",\n                extra={\n                    \"organization_id\": repo.organization_id,\n                    \"user_id\": user_id,\n                    \"repository\": repo.name,\n                    \"provider\": provider.id,\n                    \"error\": str(e),\n                    \"end_sha\": end_sha,\n                    \"start_sha\": start_sha,\n                },\n            )\n            span = sentry_sdk.get_current_span()\n            if span is None:\n                raise TypeError(\"No span is currently active right now\")\n            span.set_status(\"unknown_error\")\n            logger.exception(e)\n            if isinstance(e, InvalidIdentity) and getattr(e, \"identity\", None):\n                handle_invalid_identity(identity=e.identity, commit_failure=True)\n            elif isinstance(e, (PluginError, InvalidIdentity, IntegrationError)):\n                msg = generate_fetch_commits_error_email(release, repo, str(e))\n                emails = get_emails_for_user_or_org(user, release.organization_id)\n                msg.send_async(to=emails)\n            else:\n                msg = generate_fetch_commits_error_email(\n                    release, repo, \"An internal system error occurred.\"\n                )\n                emails = get_emails_for_user_or_org(user, release.organization_id)\n                msg.send_async(to=emails)\n        else:\n            logger.info(\n                \"fetch_commits.complete\",\n                extra={\n                    \"organization_id\": repo.organization_id,\n                    \"user_id\": user_id,\n                    \"repository\": repo.name,\n                    \"end_sha\": end_sha,\n                    \"start_sha\": start_sha,\n                    \"num_commits\": len(repo_commits or []),\n                },\n            )\n            commit_list.extend(repo_commits)\n\n    if commit_list:\n        try:\n            release.set_commits(commit_list)\n        except ReleaseCommitError:\n            # Another task or webworker is currently setting commits on this\n            # release. Return early as that task will do the remaining work.\n            logger.info(\n                \"fetch_commits.duplicate\",\n                extra={\n                    \"release_id\": release.id,\n                    \"organization_id\": release.organization_id,\n                    \"user_id\": user_id,\n                },\n            )\n            return\n\n        deploys = Deploy.objects.filter(\n            organization_id=release.organization_id, release=release, notified=False\n        ).values_list(\"id\", \"environment_id\", \"date_finished\")\n\n        # XXX(dcramer): i don't know why this would have multiple environments, but for\n        # our sanity lets assume it can\n        pending_notifications = []\n        last_deploy_per_environment = {}\n        for deploy_id, environment_id, date_finished in deploys:\n            last_deploy_per_environment[environment_id] = (deploy_id, date_finished)\n            pending_notifications.append(deploy_id)\n\n        repo_queryset = ReleaseHeadCommit.objects.filter(\n            organization_id=release.organization_id, release=release\n        ).values_list(\"repository_id\", \"commit\")\n\n        # for each repo, update (or create if this is the first one) our records\n        # of the latest commit-associated release in each env\n        # use deploys as a proxy for ReleaseEnvironment, because they contain\n        # a timestamp in addition to release and env data\n        for repository_id, commit_id in repo_queryset:\n            for environment_id, (deploy_id, date_finished) in last_deploy_per_environment.items():\n                # we need to mark LatestRepoReleaseEnvironment, but only if there's not a\n                # deploy in the given environment which has completed *after*\n                # this deploy (given we might process commits out of order)\n                if not Deploy.objects.filter(\n                    id__in=LatestRepoReleaseEnvironment.objects.filter(\n                        repository_id=repository_id, environment_id=environment_id\n                    ).values(\"deploy_id\"),\n                    date_finished__gt=date_finished,\n                ).exists():\n                    LatestRepoReleaseEnvironment.objects.create_or_update(\n                        repository_id=repository_id,\n                        environment_id=environment_id,\n                        values={\n                            \"release_id\": release.id,\n                            \"deploy_id\": deploy_id,\n                            \"commit_id\": commit_id,\n                        },\n                    )\n\n        for deploy_id in pending_notifications:\n            Deploy.notify_if_ready(deploy_id, fetch_complete=True)\n\n\ndef is_integration_provider(provider):\n    return provider and provider.startswith(\"integrations:\")\n\n\ndef get_emails_for_user_or_org(user: RpcUser | None, orgId: int):\n    emails: list[str] = []\n    if not user:\n        return []\n    if user.is_sentry_app:\n        organization = Organization.objects.get(id=orgId)\n        members = organization.get_members_with_org_roles(roles=[\"owner\"])\n        user_ids = [m.user_id for m in members if m.user_id]\n        emails = list(\n            {u.email for u in user_service.get_many(filter={\"user_ids\": user_ids}) if u.email}\n        )\n    else:\n        emails = [user.email]\n\n    return emails\n", "path": "src/sentry/tasks/commits.py"}]}