Skip to content

Commit

Permalink
fix type hints
Browse files Browse the repository at this point in the history
  • Loading branch information
troyraen committed Feb 11, 2024
1 parent 58065bb commit 62d6c47
Showing 1 changed file with 8 additions and 8 deletions.
16 changes: 8 additions & 8 deletions pittgoogle/alert.py
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,7 @@
import logging
from datetime import datetime
from pathlib import Path
from typing import TYPE_CHECKING, Any, Optional, Union
from typing import TYPE_CHECKING, Any, Dict, Optional, Union

import fastavro
from attrs import define, field
Expand Down Expand Up @@ -77,18 +77,18 @@ class Alert:
Union["google.cloud.pubsub_v1.types.PubsubMessage", types_.PubsubMessageLike]
] = field(default=None)
"""Incoming Pub/Sub message object."""
_attributes: Optional[Union[dict, "google._upb._message.ScalarMapContainer"]] = field(
_attributes: Optional[Union[Dict, "google._upb._message.ScalarMapContainer"]] = field(
default=None
)
_dict: Optional[dict] = field(default=None)
_dict: Optional[Dict] = field(default=None)
_dataframe: Optional["pd.DataFrame"] = field(default=None)
schema_name: Optional[str] = field(default=None)
_schema: Optional[types_.Schema] = field(default=None, init=False)
path: Optional[Path] = field(default=None)

# ---- class methods ---- #
@classmethod
def from_cloud_run(cls, envelope: dict, schema_name: Optional[str] = None) -> "Alert":
def from_cloud_run(cls, envelope: Dict, schema_name: Optional[str] = None) -> "Alert":
"""Create an `Alert` from an HTTP request envelope containing a Pub/Sub message, as received by a Cloud Run module.
Example code for a Cloud Run module that uses this method to open a ZTF alert:
Expand Down Expand Up @@ -148,8 +148,8 @@ def index():
@classmethod
def from_dict(
cls,
payload: dict,
attributes: Optional[Union[dict, "google._upb._message.ScalarMapContainer"]] = None,
payload: Dict,
attributes: Optional[Union[Dict, "google._upb._message.ScalarMapContainer"]] = None,
schema_name: Optional[str] = None,
) -> "Alert": # [TODO] update tom_desc to use this
"""Create an `Alert` from a dictionary (`payload`)."""
Expand All @@ -173,7 +173,7 @@ def from_path(cls, path: Union[str, Path], schema_name: Optional[str] = None) ->

# ---- properties ---- #
@property
def attributes(self) -> dict:
def attributes(self) -> Dict:
"""Custom metadata for the message. Pub/Sub handles this as a dict-like called "attributes".
If this was not set when the `Alert` was instantiated, a new dictionary will be created using
Expand All @@ -187,7 +187,7 @@ def attributes(self) -> dict:
return self._attributes

@property
def dict(self) -> dict:
def dict(self) -> Dict:
"""Alert data as a dictionary. Created from `self.msg.data`, if needed.
Raises
Expand Down

0 comments on commit 62d6c47

Please sign in to comment.