github/AutoGPT
1
1
Fork 0
mirror of https://github.com/Significant-Gravitas/AutoGPT.git synced 2026-04-08 03:00:28 -04:00
Code Issues Packages Projects Releases Wiki Activity

fix: remove create notification nonsense

Browse Source
This commit is contained in:
Nicholas Tindle
2025-02-07 20:57:07 -06:00
parent 3e9ca87fa7
commit 6e14b95e09
2 changed files with 9 additions and 71 deletions
Download Patch File Download Diff File Expand all files Collapse all files

67
autogpt_platform/backend/backend/notifications/models.py
View File

@@ -1,5 +1,5 @@
from datetime import datetime
from typing import Annotated, Generic, Literal, Optional, TypeVar, Union, overload
from typing import Annotated, Generic, Optional, TypeVar, Union
from pydantic import BaseModel, Field
@@ -108,71 +108,6 @@ class NotificationEvent(BaseModel, Generic[T_co]):
return self.type.template
# Type-safe constructors
@overload
def create_notification(
user_id: str, type: Literal[NotificationType.AGENT_RUN], data: AgentRunData
) -> NotificationEvent[AgentRunData]: ...
@overload
def create_notification(
user_id: str, type: Literal[NotificationType.ZERO_BALANCE], data: ZeroBalanceData
) -> NotificationEvent[ZeroBalanceData]: ...
@overload
def create_notification(
user_id: str, type: Literal[NotificationType.LOW_BALANCE], data: LowBalanceData
) -> NotificationEvent[LowBalanceData]: ...
@overload
def create_notification(
user_id: str,
type: Literal[NotificationType.BLOCK_EXECUTION_FAILED],
data: BlockExecutionFailedData,
) -> NotificationEvent[BlockExecutionFailedData]: ...
@overload
def create_notification(
user_id: str,
type: Literal[NotificationType.CONTINUOUS_AGENT_ERROR],
data: ContinuousAgentErrorData,
) -> NotificationEvent[ContinuousAgentErrorData]: ...
@overload
def create_notification(
user_id: str,
type: Literal[NotificationType.DAILY_SUMMARY],
data: DailySummaryData,
) -> NotificationEvent[DailySummaryData]: ...
@overload
def create_notification(
user_id: str,
type: Literal[NotificationType.WEEKLY_SUMMARY],
data: WeeklySummaryData,
) -> NotificationEvent[WeeklySummaryData]: ...
@overload
def create_notification(
user_id: str,
type: Literal[NotificationType.MONTHLY_SUMMARY],
data: MonthlySummaryData,
) -> NotificationEvent[MonthlySummaryData]: ...
def create_notification(
user_id: str, type: NotificationType, data: BaseNotificationData
) -> NotificationEvent[BaseNotificationData]:
return NotificationEvent(user_id=user_id, type=type, data=data)
class NotificationBatch(BaseModel):
user_id: str
events: list[NotificationEvent]

13
autogpt_platform/backend/backend/notifications/summary.py
View File

@@ -2,16 +2,19 @@
import logging
from collections import defaultdict
from datetime import datetime
from typing import TYPE_CHECKING
from autogpt_libs.utils.cache import thread_cached
from backend.executor.database import DatabaseManager
if TYPE_CHECKING:
from backend.executor.database import DatabaseManager
from backend.notifications.models import (
DailySummaryData,
MonthlySummaryData,
NotificationEvent,
NotificationType,
WeeklySummaryData,
create_notification,
)
from backend.util.service import get_service_client
@@ -117,7 +120,7 @@ class SummaryManager:
if summary_type == "daily":
data = DailySummaryData(date=start_time, **stats)
type_ = NotificationType.DAILY_SUMMARY
notification = create_notification(
notification = NotificationEvent(
user_id=user_id,
type=type_,
data=data,
@@ -131,7 +134,7 @@ class SummaryManager:
**stats,
)
type_ = NotificationType.WEEKLY_SUMMARY
notification = create_notification(
notification = NotificationEvent(
user_id=user_id,
type=type_,
data=data,
@@ -141,7 +144,7 @@ class SummaryManager:
month=start_time.month, year=start_time.year, **stats
)
type_ = NotificationType.MONTHLY_SUMMARY
notification = create_notification(
notification = NotificationEvent(
user_id=user_id,
type=type_,
data=data,