We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Get hourly event count from Snuba to compare with forecast
## Objective:

The forecasting algorithm generates an hourly threshold that we should compare against. Currently we fetch the entire day's event count from Snuba. This should be the hourly event count

--- 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/issues/escalating.py`
Content:
```
1 """This module has the logic for querying Snuba for the hourly event count for a list of groups.
2 This is later used for generating group forecasts for determining when a group may be escalating.
3 """
4 
5 import logging
6 from collections import defaultdict
7 from datetime import datetime, timedelta
8 from typing import Dict, List, Sequence, Tuple, TypedDict
9 
10 from snuba_sdk import (
11     Column,
12     Condition,
13     Direction,
14     Entity,
15     Function,
16     Limit,
17     Offset,
18     Op,
19     OrderBy,
20     Query,
21     Request,
22 )
23 
24 from sentry.issues.escalating_group_forecast import EscalatingGroupForecast
25 from sentry.issues.escalating_issues_alg import GroupCount
26 from sentry.models import Group
27 from sentry.models.group import GroupStatus
28 from sentry.models.groupinbox import GroupInboxReason, add_group_to_inbox
29 from sentry.snuba.dataset import Dataset, EntityKey
30 from sentry.types.group import GroupSubStatus
31 from sentry.utils.cache import cache
32 from sentry.utils.snuba import raw_snql_query
33 
34 logger = logging.getLogger(__name__)
35 
36 __all__ = ["query_groups_past_counts", "parse_groups_past_counts"]
37 
38 REFERRER = "sentry.issues.escalating"
39 ELEMENTS_PER_SNUBA_PAGE = 10000  # This is the maximum value for Snuba
40 # The amount of data needed to generate a group forecast
41 BUCKETS_PER_GROUP = 7 * 24
42 ONE_WEEK_DURATION = 7
43 IS_ESCALATING_REFERRER = "sentry.issues.escalating.is_escalating"
44 GROUP_DAILY_COUNT_TTL = 60
45 
46 GroupsCountResponse = TypedDict(
47     "GroupsCountResponse",
48     {"group_id": int, "hourBucket": str, "count()": int, "project_id": int},
49 )
50 
51 ParsedGroupsCount = Dict[int, GroupCount]
52 
53 
54 def query_groups_past_counts(groups: Sequence[Group]) -> List[GroupsCountResponse]:
55     """Query Snuba for the counts for every group bucketed into hours.
56 
57     It optimizes the query by guaranteeing that we look at group_ids that are from the same project id.
58     This is important for Snuba as the data is stored in blocks related to the project id.
59 
60     We maximize the number of projects and groups to reduce the total number of Snuba queries.
61     Each project may not have enough groups in order to reach the max number of returned
62     elements (ELEMENTS_PER_SNUBA_PAGE), thus, projects with few groups should be grouped together until
63     we get at least a certain number of groups.
64 
65     NOTE: Groups with less than the maximum number of buckets (think of groups with just 1 event or less
66     than 7 days old) will skew the optimization since we may only get one page and less elements than the max
67     ELEMENTS_PER_SNUBA_PAGE.
68     """
69     all_results = []  # type: ignore[var-annotated]
70     if not groups:
71         return all_results
72 
73     start_date, end_date = _start_and_end_dates()
74     group_ids_by_project = _extract_project_and_group_ids(groups)
75     proj_ids, group_ids = [], []
76     processed_projects = 0
77     total_projects_count = len(group_ids_by_project)
78     organization_id = groups[0].project.organization.id
79 
80     # This iteration guarantees that all groups for a project will be queried in the same call
81     # and only one page where the groups could be mixed with groups from another project
82     # Iterating over the sorted keys guarantees results for tests
83     for proj_id in sorted(group_ids_by_project.keys()):
84         _group_ids = group_ids_by_project[proj_id]
85         # Add them to the list of projects and groups to query
86         proj_ids.append(proj_id)
87         group_ids += _group_ids
88         processed_projects += 1
89         potential_num_elements = len(_group_ids) * BUCKETS_PER_GROUP
90         # This is trying to maximize the number of groups on the first page
91         if (
92             processed_projects < total_projects_count
93             and potential_num_elements < ELEMENTS_PER_SNUBA_PAGE
94         ):
95             continue
96 
97         # TODO: Write this as a dispatcher type task and fire off a separate task per proj_ids
98         all_results += _query_with_pagination(
99             organization_id, proj_ids, group_ids, start_date, end_date
100         )
101         # We're ready for a new set of projects and ids
102         proj_ids, group_ids = [], []
103 
104     return all_results
105 
106 
107 def _query_with_pagination(
108     organization_id: int,
109     project_ids: Sequence[int],
110     group_ids: Sequence[int],
111     start_date: datetime,
112     end_date: datetime,
113 ) -> List[GroupsCountResponse]:
114     """Query Snuba for event counts for the given list of project ids and groups ids in
115     a time range."""
116     all_results = []
117     offset = 0
118     while True:
119         query = _generate_query(project_ids, group_ids, offset, start_date, end_date)
120         request = Request(
121             dataset=Dataset.Events.value,
122             app_id=REFERRER,
123             query=query,
124             tenant_ids={"referrer": REFERRER, "organization_id": organization_id},
125         )
126         results = raw_snql_query(request, referrer=REFERRER)["data"]
127         all_results += results
128         offset += ELEMENTS_PER_SNUBA_PAGE
129         if not results or len(results) < ELEMENTS_PER_SNUBA_PAGE:
130             break
131 
132     return all_results
133 
134 
135 def parse_groups_past_counts(response: Sequence[GroupsCountResponse]) -> ParsedGroupsCount:
136     """
137     Return the parsed snuba response for groups past counts to be used in generate_issue_forecast.
138     ParsedGroupCount is of the form {<group_id>: {"intervals": [str], "data": [int]}}.
139 
140     `response`: Snuba response for group event counts
141     """
142     group_counts: ParsedGroupsCount = {}
143     group_ids_list = group_counts.keys()
144     for data in response:
145         group_id = data["group_id"]
146         if group_id not in group_ids_list:
147             group_counts[group_id] = {
148                 "intervals": [data["hourBucket"]],
149                 "data": [data["count()"]],
150             }
151         else:
152             group_counts[group_id]["intervals"].append(data["hourBucket"])
153             group_counts[group_id]["data"].append(data["count()"])
154     return group_counts
155 
156 
157 def _generate_query(
158     project_ids: Sequence[int],
159     group_ids: Sequence[int],
160     offset: int,
161     start_date: datetime,
162     end_date: datetime,
163 ) -> Query:
164     """This simply generates a query based on the passed parameters"""
165     group_id_col = Column("group_id")
166     proj_id_col = Column("project_id")
167     return Query(
168         match=Entity(EntityKey.Events.value),
169         select=[
170             proj_id_col,
171             group_id_col,
172             Function("toStartOfHour", [Column("timestamp")], "hourBucket"),
173             Function("count", []),
174         ],
175         groupby=[proj_id_col, group_id_col, Column("hourBucket")],
176         where=[
177             Condition(proj_id_col, Op.IN, Function("tuple", project_ids)),
178             Condition(Column("group_id"), Op.IN, Function("tuple", group_ids)),
179             Condition(Column("timestamp"), Op.GTE, start_date),
180             Condition(Column("timestamp"), Op.LT, end_date),
181         ],
182         limit=Limit(ELEMENTS_PER_SNUBA_PAGE),
183         offset=Offset(offset),
184         orderby=[
185             OrderBy(proj_id_col, Direction.ASC),
186             OrderBy(group_id_col, Direction.ASC),
187             OrderBy(Column("hourBucket"), Direction.ASC),
188         ],
189     )
190 
191 
192 def _start_and_end_dates(hours: int = BUCKETS_PER_GROUP) -> Tuple[datetime, datetime]:
193     """Return the start and end date of N hours time range."""
194     end_datetime = datetime.now()
195     return end_datetime - timedelta(hours=hours), end_datetime
196 
197 
198 def _extract_project_and_group_ids(groups: Sequence[Group]) -> Dict[int, List[int]]:
199     """Return all project and group IDs from a list of Group"""
200     group_ids_by_project: Dict[int, List[int]] = defaultdict(list)
201     for group in groups:
202         group_ids_by_project[group.project_id].append(group.id)
203 
204     return group_ids_by_project
205 
206 
207 def get_group_daily_count(organization_id: int, project_id: int, group_id: int) -> int:
208     """Return the number of events a group has had today"""
209     key = f"daily-group-count:{project_id}:{group_id}"
210     daily_count = cache.get(key)
211 
212     if daily_count is None:
213         today = datetime.now().date()
214         midnight = datetime.combine(today, datetime.min.time())
215         now = datetime.now()
216         query = Query(
217             match=Entity(EntityKey.Events.value),
218             select=[
219                 Function("count", []),
220             ],
221             where=[
222                 Condition(Column("project_id"), Op.EQ, project_id),
223                 Condition(Column("group_id"), Op.EQ, group_id),
224                 Condition(Column("timestamp"), Op.GTE, midnight),
225                 Condition(Column("timestamp"), Op.LT, now),
226             ],
227         )
228         request = Request(
229             dataset=Dataset.Events.value,
230             app_id=IS_ESCALATING_REFERRER,
231             query=query,
232             tenant_ids={"referrer": IS_ESCALATING_REFERRER, "organization_id": organization_id},
233         )
234         daily_count = int(
235             raw_snql_query(request, referrer=IS_ESCALATING_REFERRER)["data"][0]["count()"]
236         )
237         cache.set(key, daily_count, GROUP_DAILY_COUNT_TTL)
238     return int(daily_count)
239 
240 
241 def is_escalating(group: Group) -> bool:
242     """Return boolean depending on if the group is escalating or not"""
243     group_daily_count = get_group_daily_count(
244         group.project.organization.id, group.project.id, group.id
245     )
246     forecast_today = EscalatingGroupForecast.fetch_todays_forecast(group.project.id, group.id)
247     # Check if current event occurance is greater than forecast for today's date
248     if group_daily_count > forecast_today:
249         group.substatus = GroupSubStatus.ESCALATING
250         group.status = GroupStatus.UNRESOLVED
251         add_group_to_inbox(group, GroupInboxReason.ESCALATING)
252         return True
253     return False
254
```
--- 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/issues/escalating.py b/src/sentry/issues/escalating.py
--- a/src/sentry/issues/escalating.py
+++ b/src/sentry/issues/escalating.py
@@ -41,7 +41,7 @@
 BUCKETS_PER_GROUP = 7 * 24
 ONE_WEEK_DURATION = 7
 IS_ESCALATING_REFERRER = "sentry.issues.escalating.is_escalating"
-GROUP_DAILY_COUNT_TTL = 60
+GROUP_HOURLY_COUNT_TTL = 60
 
 GroupsCountResponse = TypedDict(
     "GroupsCountResponse",
@@ -204,15 +204,14 @@
     return group_ids_by_project
 
 
-def get_group_daily_count(organization_id: int, project_id: int, group_id: int) -> int:
-    """Return the number of events a group has had today"""
-    key = f"daily-group-count:{project_id}:{group_id}"
-    daily_count = cache.get(key)
+def get_group_hourly_count(organization_id: int, project_id: int, group_id: int) -> int:
+    """Return the number of events a group has had today in the last hour"""
+    key = f"hourly-group-count:{project_id}:{group_id}"
+    hourly_count = cache.get(key)
 
-    if daily_count is None:
-        today = datetime.now().date()
-        midnight = datetime.combine(today, datetime.min.time())
+    if hourly_count is None:
         now = datetime.now()
+        current_hour = now.replace(minute=0, second=0, microsecond=0)
         query = Query(
             match=Entity(EntityKey.Events.value),
             select=[
@@ -221,7 +220,7 @@
             where=[
                 Condition(Column("project_id"), Op.EQ, project_id),
                 Condition(Column("group_id"), Op.EQ, group_id),
-                Condition(Column("timestamp"), Op.GTE, midnight),
+                Condition(Column("timestamp"), Op.GTE, current_hour),
                 Condition(Column("timestamp"), Op.LT, now),
             ],
         )
@@ -231,21 +230,21 @@
             query=query,
             tenant_ids={"referrer": IS_ESCALATING_REFERRER, "organization_id": organization_id},
         )
-        daily_count = int(
+        hourly_count = int(
             raw_snql_query(request, referrer=IS_ESCALATING_REFERRER)["data"][0]["count()"]
         )
-        cache.set(key, daily_count, GROUP_DAILY_COUNT_TTL)
-    return int(daily_count)
+        cache.set(key, hourly_count, GROUP_HOURLY_COUNT_TTL)
+    return int(hourly_count)
 
 
 def is_escalating(group: Group) -> bool:
     """Return boolean depending on if the group is escalating or not"""
-    group_daily_count = get_group_daily_count(
+    group_hourly_count = get_group_hourly_count(
         group.project.organization.id, group.project.id, group.id
     )
     forecast_today = EscalatingGroupForecast.fetch_todays_forecast(group.project.id, group.id)
     # Check if current event occurance is greater than forecast for today's date
-    if group_daily_count > forecast_today:
+    if group_hourly_count > forecast_today:
         group.substatus = GroupSubStatus.ESCALATING
         group.status = GroupStatus.UNRESOLVED
         add_group_to_inbox(group, GroupInboxReason.ESCALATING)

{"golden_diff": "diff --git a/src/sentry/issues/escalating.py b/src/sentry/issues/escalating.py\n--- a/src/sentry/issues/escalating.py\n+++ b/src/sentry/issues/escalating.py\n@@ -41,7 +41,7 @@\n BUCKETS_PER_GROUP = 7 * 24\n ONE_WEEK_DURATION = 7\n IS_ESCALATING_REFERRER = \"sentry.issues.escalating.is_escalating\"\n-GROUP_DAILY_COUNT_TTL = 60\n+GROUP_HOURLY_COUNT_TTL = 60\n \n GroupsCountResponse = TypedDict(\n     \"GroupsCountResponse\",\n@@ -204,15 +204,14 @@\n     return group_ids_by_project\n \n \n-def get_group_daily_count(organization_id: int, project_id: int, group_id: int) -> int:\n-    \"\"\"Return the number of events a group has had today\"\"\"\n-    key = f\"daily-group-count:{project_id}:{group_id}\"\n-    daily_count = cache.get(key)\n+def get_group_hourly_count(organization_id: int, project_id: int, group_id: int) -> int:\n+    \"\"\"Return the number of events a group has had today in the last hour\"\"\"\n+    key = f\"hourly-group-count:{project_id}:{group_id}\"\n+    hourly_count = cache.get(key)\n \n-    if daily_count is None:\n-        today = datetime.now().date()\n-        midnight = datetime.combine(today, datetime.min.time())\n+    if hourly_count is None:\n         now = datetime.now()\n+        current_hour = now.replace(minute=0, second=0, microsecond=0)\n         query = Query(\n             match=Entity(EntityKey.Events.value),\n             select=[\n@@ -221,7 +220,7 @@\n             where=[\n                 Condition(Column(\"project_id\"), Op.EQ, project_id),\n                 Condition(Column(\"group_id\"), Op.EQ, group_id),\n-                Condition(Column(\"timestamp\"), Op.GTE, midnight),\n+                Condition(Column(\"timestamp\"), Op.GTE, current_hour),\n                 Condition(Column(\"timestamp\"), Op.LT, now),\n             ],\n         )\n@@ -231,21 +230,21 @@\n             query=query,\n             tenant_ids={\"referrer\": IS_ESCALATING_REFERRER, \"organization_id\": organization_id},\n         )\n-        daily_count = int(\n+        hourly_count = int(\n             raw_snql_query(request, referrer=IS_ESCALATING_REFERRER)[\"data\"][0][\"count()\"]\n         )\n-        cache.set(key, daily_count, GROUP_DAILY_COUNT_TTL)\n-    return int(daily_count)\n+        cache.set(key, hourly_count, GROUP_HOURLY_COUNT_TTL)\n+    return int(hourly_count)\n \n \n def is_escalating(group: Group) -> bool:\n     \"\"\"Return boolean depending on if the group is escalating or not\"\"\"\n-    group_daily_count = get_group_daily_count(\n+    group_hourly_count = get_group_hourly_count(\n         group.project.organization.id, group.project.id, group.id\n     )\n     forecast_today = EscalatingGroupForecast.fetch_todays_forecast(group.project.id, group.id)\n     # Check if current event occurance is greater than forecast for today's date\n-    if group_daily_count > forecast_today:\n+    if group_hourly_count > forecast_today:\n         group.substatus = GroupSubStatus.ESCALATING\n         group.status = GroupStatus.UNRESOLVED\n         add_group_to_inbox(group, GroupInboxReason.ESCALATING)\n", "issue": "Get hourly event count from Snuba to compare with forecast\n## Objective:\n\nThe forecasting algorithm generates an hourly threshold that we should compare against. Currently we fetch the entire day's event count from Snuba. This should be the hourly event count\n", "before_files": [{"content": "\"\"\"This module has the logic for querying Snuba for the hourly event count for a list of groups.\nThis is later used for generating group forecasts for determining when a group may be escalating.\n\"\"\"\n\nimport logging\nfrom collections import defaultdict\nfrom datetime import datetime, timedelta\nfrom typing import Dict, List, Sequence, Tuple, TypedDict\n\nfrom snuba_sdk import (\n    Column,\n    Condition,\n    Direction,\n    Entity,\n    Function,\n    Limit,\n    Offset,\n    Op,\n    OrderBy,\n    Query,\n    Request,\n)\n\nfrom sentry.issues.escalating_group_forecast import EscalatingGroupForecast\nfrom sentry.issues.escalating_issues_alg import GroupCount\nfrom sentry.models import Group\nfrom sentry.models.group import GroupStatus\nfrom sentry.models.groupinbox import GroupInboxReason, add_group_to_inbox\nfrom sentry.snuba.dataset import Dataset, EntityKey\nfrom sentry.types.group import GroupSubStatus\nfrom sentry.utils.cache import cache\nfrom sentry.utils.snuba import raw_snql_query\n\nlogger = logging.getLogger(__name__)\n\n__all__ = [\"query_groups_past_counts\", \"parse_groups_past_counts\"]\n\nREFERRER = \"sentry.issues.escalating\"\nELEMENTS_PER_SNUBA_PAGE = 10000  # This is the maximum value for Snuba\n# The amount of data needed to generate a group forecast\nBUCKETS_PER_GROUP = 7 * 24\nONE_WEEK_DURATION = 7\nIS_ESCALATING_REFERRER = \"sentry.issues.escalating.is_escalating\"\nGROUP_DAILY_COUNT_TTL = 60\n\nGroupsCountResponse = TypedDict(\n    \"GroupsCountResponse\",\n    {\"group_id\": int, \"hourBucket\": str, \"count()\": int, \"project_id\": int},\n)\n\nParsedGroupsCount = Dict[int, GroupCount]\n\n\ndef query_groups_past_counts(groups: Sequence[Group]) -> List[GroupsCountResponse]:\n    \"\"\"Query Snuba for the counts for every group bucketed into hours.\n\n    It optimizes the query by guaranteeing that we look at group_ids that are from the same project id.\n    This is important for Snuba as the data is stored in blocks related to the project id.\n\n    We maximize the number of projects and groups to reduce the total number of Snuba queries.\n    Each project may not have enough groups in order to reach the max number of returned\n    elements (ELEMENTS_PER_SNUBA_PAGE), thus, projects with few groups should be grouped together until\n    we get at least a certain number of groups.\n\n    NOTE: Groups with less than the maximum number of buckets (think of groups with just 1 event or less\n    than 7 days old) will skew the optimization since we may only get one page and less elements than the max\n    ELEMENTS_PER_SNUBA_PAGE.\n    \"\"\"\n    all_results = []  # type: ignore[var-annotated]\n    if not groups:\n        return all_results\n\n    start_date, end_date = _start_and_end_dates()\n    group_ids_by_project = _extract_project_and_group_ids(groups)\n    proj_ids, group_ids = [], []\n    processed_projects = 0\n    total_projects_count = len(group_ids_by_project)\n    organization_id = groups[0].project.organization.id\n\n    # This iteration guarantees that all groups for a project will be queried in the same call\n    # and only one page where the groups could be mixed with groups from another project\n    # Iterating over the sorted keys guarantees results for tests\n    for proj_id in sorted(group_ids_by_project.keys()):\n        _group_ids = group_ids_by_project[proj_id]\n        # Add them to the list of projects and groups to query\n        proj_ids.append(proj_id)\n        group_ids += _group_ids\n        processed_projects += 1\n        potential_num_elements = len(_group_ids) * BUCKETS_PER_GROUP\n        # This is trying to maximize the number of groups on the first page\n        if (\n            processed_projects < total_projects_count\n            and potential_num_elements < ELEMENTS_PER_SNUBA_PAGE\n        ):\n            continue\n\n        # TODO: Write this as a dispatcher type task and fire off a separate task per proj_ids\n        all_results += _query_with_pagination(\n            organization_id, proj_ids, group_ids, start_date, end_date\n        )\n        # We're ready for a new set of projects and ids\n        proj_ids, group_ids = [], []\n\n    return all_results\n\n\ndef _query_with_pagination(\n    organization_id: int,\n    project_ids: Sequence[int],\n    group_ids: Sequence[int],\n    start_date: datetime,\n    end_date: datetime,\n) -> List[GroupsCountResponse]:\n    \"\"\"Query Snuba for event counts for the given list of project ids and groups ids in\n    a time range.\"\"\"\n    all_results = []\n    offset = 0\n    while True:\n        query = _generate_query(project_ids, group_ids, offset, start_date, end_date)\n        request = Request(\n            dataset=Dataset.Events.value,\n            app_id=REFERRER,\n            query=query,\n            tenant_ids={\"referrer\": REFERRER, \"organization_id\": organization_id},\n        )\n        results = raw_snql_query(request, referrer=REFERRER)[\"data\"]\n        all_results += results\n        offset += ELEMENTS_PER_SNUBA_PAGE\n        if not results or len(results) < ELEMENTS_PER_SNUBA_PAGE:\n            break\n\n    return all_results\n\n\ndef parse_groups_past_counts(response: Sequence[GroupsCountResponse]) -> ParsedGroupsCount:\n    \"\"\"\n    Return the parsed snuba response for groups past counts to be used in generate_issue_forecast.\n    ParsedGroupCount is of the form {<group_id>: {\"intervals\": [str], \"data\": [int]}}.\n\n    `response`: Snuba response for group event counts\n    \"\"\"\n    group_counts: ParsedGroupsCount = {}\n    group_ids_list = group_counts.keys()\n    for data in response:\n        group_id = data[\"group_id\"]\n        if group_id not in group_ids_list:\n            group_counts[group_id] = {\n                \"intervals\": [data[\"hourBucket\"]],\n                \"data\": [data[\"count()\"]],\n            }\n        else:\n            group_counts[group_id][\"intervals\"].append(data[\"hourBucket\"])\n            group_counts[group_id][\"data\"].append(data[\"count()\"])\n    return group_counts\n\n\ndef _generate_query(\n    project_ids: Sequence[int],\n    group_ids: Sequence[int],\n    offset: int,\n    start_date: datetime,\n    end_date: datetime,\n) -> Query:\n    \"\"\"This simply generates a query based on the passed parameters\"\"\"\n    group_id_col = Column(\"group_id\")\n    proj_id_col = Column(\"project_id\")\n    return Query(\n        match=Entity(EntityKey.Events.value),\n        select=[\n            proj_id_col,\n            group_id_col,\n            Function(\"toStartOfHour\", [Column(\"timestamp\")], \"hourBucket\"),\n            Function(\"count\", []),\n        ],\n        groupby=[proj_id_col, group_id_col, Column(\"hourBucket\")],\n        where=[\n            Condition(proj_id_col, Op.IN, Function(\"tuple\", project_ids)),\n            Condition(Column(\"group_id\"), Op.IN, Function(\"tuple\", group_ids)),\n            Condition(Column(\"timestamp\"), Op.GTE, start_date),\n            Condition(Column(\"timestamp\"), Op.LT, end_date),\n        ],\n        limit=Limit(ELEMENTS_PER_SNUBA_PAGE),\n        offset=Offset(offset),\n        orderby=[\n            OrderBy(proj_id_col, Direction.ASC),\n            OrderBy(group_id_col, Direction.ASC),\n            OrderBy(Column(\"hourBucket\"), Direction.ASC),\n        ],\n    )\n\n\ndef _start_and_end_dates(hours: int = BUCKETS_PER_GROUP) -> Tuple[datetime, datetime]:\n    \"\"\"Return the start and end date of N hours time range.\"\"\"\n    end_datetime = datetime.now()\n    return end_datetime - timedelta(hours=hours), end_datetime\n\n\ndef _extract_project_and_group_ids(groups: Sequence[Group]) -> Dict[int, List[int]]:\n    \"\"\"Return all project and group IDs from a list of Group\"\"\"\n    group_ids_by_project: Dict[int, List[int]] = defaultdict(list)\n    for group in groups:\n        group_ids_by_project[group.project_id].append(group.id)\n\n    return group_ids_by_project\n\n\ndef get_group_daily_count(organization_id: int, project_id: int, group_id: int) -> int:\n    \"\"\"Return the number of events a group has had today\"\"\"\n    key = f\"daily-group-count:{project_id}:{group_id}\"\n    daily_count = cache.get(key)\n\n    if daily_count is None:\n        today = datetime.now().date()\n        midnight = datetime.combine(today, datetime.min.time())\n        now = datetime.now()\n        query = Query(\n            match=Entity(EntityKey.Events.value),\n            select=[\n                Function(\"count\", []),\n            ],\n            where=[\n                Condition(Column(\"project_id\"), Op.EQ, project_id),\n                Condition(Column(\"group_id\"), Op.EQ, group_id),\n                Condition(Column(\"timestamp\"), Op.GTE, midnight),\n                Condition(Column(\"timestamp\"), Op.LT, now),\n            ],\n        )\n        request = Request(\n            dataset=Dataset.Events.value,\n            app_id=IS_ESCALATING_REFERRER,\n            query=query,\n            tenant_ids={\"referrer\": IS_ESCALATING_REFERRER, \"organization_id\": organization_id},\n        )\n        daily_count = int(\n            raw_snql_query(request, referrer=IS_ESCALATING_REFERRER)[\"data\"][0][\"count()\"]\n        )\n        cache.set(key, daily_count, GROUP_DAILY_COUNT_TTL)\n    return int(daily_count)\n\n\ndef is_escalating(group: Group) -> bool:\n    \"\"\"Return boolean depending on if the group is escalating or not\"\"\"\n    group_daily_count = get_group_daily_count(\n        group.project.organization.id, group.project.id, group.id\n    )\n    forecast_today = EscalatingGroupForecast.fetch_todays_forecast(group.project.id, group.id)\n    # Check if current event occurance is greater than forecast for today's date\n    if group_daily_count > forecast_today:\n        group.substatus = GroupSubStatus.ESCALATING\n        group.status = GroupStatus.UNRESOLVED\n        add_group_to_inbox(group, GroupInboxReason.ESCALATING)\n        return True\n    return False\n", "path": "src/sentry/issues/escalating.py"}], "after_files": [{"content": "\"\"\"This module has the logic for querying Snuba for the hourly event count for a list of groups.\nThis is later used for generating group forecasts for determining when a group may be escalating.\n\"\"\"\n\nimport logging\nfrom collections import defaultdict\nfrom datetime import datetime, timedelta\nfrom typing import Dict, List, Sequence, Tuple, TypedDict\n\nfrom snuba_sdk import (\n    Column,\n    Condition,\n    Direction,\n    Entity,\n    Function,\n    Limit,\n    Offset,\n    Op,\n    OrderBy,\n    Query,\n    Request,\n)\n\nfrom sentry.issues.escalating_group_forecast import EscalatingGroupForecast\nfrom sentry.issues.escalating_issues_alg import GroupCount\nfrom sentry.models import Group\nfrom sentry.models.group import GroupStatus\nfrom sentry.models.groupinbox import GroupInboxReason, add_group_to_inbox\nfrom sentry.snuba.dataset import Dataset, EntityKey\nfrom sentry.types.group import GroupSubStatus\nfrom sentry.utils.cache import cache\nfrom sentry.utils.snuba import raw_snql_query\n\nlogger = logging.getLogger(__name__)\n\n__all__ = [\"query_groups_past_counts\", \"parse_groups_past_counts\"]\n\nREFERRER = \"sentry.issues.escalating\"\nELEMENTS_PER_SNUBA_PAGE = 10000  # This is the maximum value for Snuba\n# The amount of data needed to generate a group forecast\nBUCKETS_PER_GROUP = 7 * 24\nONE_WEEK_DURATION = 7\nIS_ESCALATING_REFERRER = \"sentry.issues.escalating.is_escalating\"\nGROUP_HOURLY_COUNT_TTL = 60\n\nGroupsCountResponse = TypedDict(\n    \"GroupsCountResponse\",\n    {\"group_id\": int, \"hourBucket\": str, \"count()\": int, \"project_id\": int},\n)\n\nParsedGroupsCount = Dict[int, GroupCount]\n\n\ndef query_groups_past_counts(groups: Sequence[Group]) -> List[GroupsCountResponse]:\n    \"\"\"Query Snuba for the counts for every group bucketed into hours.\n\n    It optimizes the query by guaranteeing that we look at group_ids that are from the same project id.\n    This is important for Snuba as the data is stored in blocks related to the project id.\n\n    We maximize the number of projects and groups to reduce the total number of Snuba queries.\n    Each project may not have enough groups in order to reach the max number of returned\n    elements (ELEMENTS_PER_SNUBA_PAGE), thus, projects with few groups should be grouped together until\n    we get at least a certain number of groups.\n\n    NOTE: Groups with less than the maximum number of buckets (think of groups with just 1 event or less\n    than 7 days old) will skew the optimization since we may only get one page and less elements than the max\n    ELEMENTS_PER_SNUBA_PAGE.\n    \"\"\"\n    all_results = []  # type: ignore[var-annotated]\n    if not groups:\n        return all_results\n\n    start_date, end_date = _start_and_end_dates()\n    group_ids_by_project = _extract_project_and_group_ids(groups)\n    proj_ids, group_ids = [], []\n    processed_projects = 0\n    total_projects_count = len(group_ids_by_project)\n    organization_id = groups[0].project.organization.id\n\n    # This iteration guarantees that all groups for a project will be queried in the same call\n    # and only one page where the groups could be mixed with groups from another project\n    # Iterating over the sorted keys guarantees results for tests\n    for proj_id in sorted(group_ids_by_project.keys()):\n        _group_ids = group_ids_by_project[proj_id]\n        # Add them to the list of projects and groups to query\n        proj_ids.append(proj_id)\n        group_ids += _group_ids\n        processed_projects += 1\n        potential_num_elements = len(_group_ids) * BUCKETS_PER_GROUP\n        # This is trying to maximize the number of groups on the first page\n        if (\n            processed_projects < total_projects_count\n            and potential_num_elements < ELEMENTS_PER_SNUBA_PAGE\n        ):\n            continue\n\n        # TODO: Write this as a dispatcher type task and fire off a separate task per proj_ids\n        all_results += _query_with_pagination(\n            organization_id, proj_ids, group_ids, start_date, end_date\n        )\n        # We're ready for a new set of projects and ids\n        proj_ids, group_ids = [], []\n\n    return all_results\n\n\ndef _query_with_pagination(\n    organization_id: int,\n    project_ids: Sequence[int],\n    group_ids: Sequence[int],\n    start_date: datetime,\n    end_date: datetime,\n) -> List[GroupsCountResponse]:\n    \"\"\"Query Snuba for event counts for the given list of project ids and groups ids in\n    a time range.\"\"\"\n    all_results = []\n    offset = 0\n    while True:\n        query = _generate_query(project_ids, group_ids, offset, start_date, end_date)\n        request = Request(\n            dataset=Dataset.Events.value,\n            app_id=REFERRER,\n            query=query,\n            tenant_ids={\"referrer\": REFERRER, \"organization_id\": organization_id},\n        )\n        results = raw_snql_query(request, referrer=REFERRER)[\"data\"]\n        all_results += results\n        offset += ELEMENTS_PER_SNUBA_PAGE\n        if not results or len(results) < ELEMENTS_PER_SNUBA_PAGE:\n            break\n\n    return all_results\n\n\ndef parse_groups_past_counts(response: Sequence[GroupsCountResponse]) -> ParsedGroupsCount:\n    \"\"\"\n    Return the parsed snuba response for groups past counts to be used in generate_issue_forecast.\n    ParsedGroupCount is of the form {<group_id>: {\"intervals\": [str], \"data\": [int]}}.\n\n    `response`: Snuba response for group event counts\n    \"\"\"\n    group_counts: ParsedGroupsCount = {}\n    group_ids_list = group_counts.keys()\n    for data in response:\n        group_id = data[\"group_id\"]\n        if group_id not in group_ids_list:\n            group_counts[group_id] = {\n                \"intervals\": [data[\"hourBucket\"]],\n                \"data\": [data[\"count()\"]],\n            }\n        else:\n            group_counts[group_id][\"intervals\"].append(data[\"hourBucket\"])\n            group_counts[group_id][\"data\"].append(data[\"count()\"])\n    return group_counts\n\n\ndef _generate_query(\n    project_ids: Sequence[int],\n    group_ids: Sequence[int],\n    offset: int,\n    start_date: datetime,\n    end_date: datetime,\n) -> Query:\n    \"\"\"This simply generates a query based on the passed parameters\"\"\"\n    group_id_col = Column(\"group_id\")\n    proj_id_col = Column(\"project_id\")\n    return Query(\n        match=Entity(EntityKey.Events.value),\n        select=[\n            proj_id_col,\n            group_id_col,\n            Function(\"toStartOfHour\", [Column(\"timestamp\")], \"hourBucket\"),\n            Function(\"count\", []),\n        ],\n        groupby=[proj_id_col, group_id_col, Column(\"hourBucket\")],\n        where=[\n            Condition(proj_id_col, Op.IN, Function(\"tuple\", project_ids)),\n            Condition(Column(\"group_id\"), Op.IN, Function(\"tuple\", group_ids)),\n            Condition(Column(\"timestamp\"), Op.GTE, start_date),\n            Condition(Column(\"timestamp\"), Op.LT, end_date),\n        ],\n        limit=Limit(ELEMENTS_PER_SNUBA_PAGE),\n        offset=Offset(offset),\n        orderby=[\n            OrderBy(proj_id_col, Direction.ASC),\n            OrderBy(group_id_col, Direction.ASC),\n            OrderBy(Column(\"hourBucket\"), Direction.ASC),\n        ],\n    )\n\n\ndef _start_and_end_dates(hours: int = BUCKETS_PER_GROUP) -> Tuple[datetime, datetime]:\n    \"\"\"Return the start and end date of N hours time range.\"\"\"\n    end_datetime = datetime.now()\n    return end_datetime - timedelta(hours=hours), end_datetime\n\n\ndef _extract_project_and_group_ids(groups: Sequence[Group]) -> Dict[int, List[int]]:\n    \"\"\"Return all project and group IDs from a list of Group\"\"\"\n    group_ids_by_project: Dict[int, List[int]] = defaultdict(list)\n    for group in groups:\n        group_ids_by_project[group.project_id].append(group.id)\n\n    return group_ids_by_project\n\n\ndef get_group_hourly_count(organization_id: int, project_id: int, group_id: int) -> int:\n    \"\"\"Return the number of events a group has had today in the last hour\"\"\"\n    key = f\"hourly-group-count:{project_id}:{group_id}\"\n    hourly_count = cache.get(key)\n\n    if hourly_count is None:\n        now = datetime.now()\n        current_hour = now.replace(minute=0, second=0, microsecond=0)\n        query = Query(\n            match=Entity(EntityKey.Events.value),\n            select=[\n                Function(\"count\", []),\n            ],\n            where=[\n                Condition(Column(\"project_id\"), Op.EQ, project_id),\n                Condition(Column(\"group_id\"), Op.EQ, group_id),\n                Condition(Column(\"timestamp\"), Op.GTE, current_hour),\n                Condition(Column(\"timestamp\"), Op.LT, now),\n            ],\n        )\n        request = Request(\n            dataset=Dataset.Events.value,\n            app_id=IS_ESCALATING_REFERRER,\n            query=query,\n            tenant_ids={\"referrer\": IS_ESCALATING_REFERRER, \"organization_id\": organization_id},\n        )\n        hourly_count = int(\n            raw_snql_query(request, referrer=IS_ESCALATING_REFERRER)[\"data\"][0][\"count()\"]\n        )\n        cache.set(key, hourly_count, GROUP_HOURLY_COUNT_TTL)\n    return int(hourly_count)\n\n\ndef is_escalating(group: Group) -> bool:\n    \"\"\"Return boolean depending on if the group is escalating or not\"\"\"\n    group_hourly_count = get_group_hourly_count(\n        group.project.organization.id, group.project.id, group.id\n    )\n    forecast_today = EscalatingGroupForecast.fetch_todays_forecast(group.project.id, group.id)\n    # Check if current event occurance is greater than forecast for today's date\n    if group_hourly_count > forecast_today:\n        group.substatus = GroupSubStatus.ESCALATING\n        group.status = GroupStatus.UNRESOLVED\n        add_group_to_inbox(group, GroupInboxReason.ESCALATING)\n        return True\n    return False\n", "path": "src/sentry/issues/escalating.py"}]}