Skip to content

Commit

Permalink
ref(typing): Switch from type comments to annotations
Browse files Browse the repository at this point in the history
  • Loading branch information
sentrivana committed Jan 26, 2024
1 parent d600037 commit b626b17
Show file tree
Hide file tree
Showing 103 changed files with 2,595 additions and 3,362 deletions.
9 changes: 3 additions & 6 deletions sentry_sdk/_compat.py
Expand Up @@ -14,18 +14,15 @@
PY311 = sys.version_info[0] == 3 and sys.version_info[1] >= 11


def with_metaclass(meta, *bases):
# type: (Any, *Any) -> Any
def with_metaclass(meta: Any, *bases: Any) -> Any:
class MetaClass(type):
def __new__(metacls, name, this_bases, d):
# type: (Any, Any, Any, Any) -> Any
def __new__(metacls: Any, name: Any, this_bases: Any, d: Any) -> Any:
return meta(name, bases, d)

return type.__new__(MetaClass, "temporary_class", (), {})


def check_thread_support():
# type: () -> None
def check_thread_support() -> None:
try:
from uwsgi import opt # type: ignore
except ImportError:
Expand Down
2 changes: 1 addition & 1 deletion sentry_sdk/_queue.py
Expand Up @@ -273,7 +273,7 @@ def get_nowait(self):

# Initialize the queue representation
def _init(self, maxsize):
self.queue = deque() # type: Any
self.queue: Any = deque()

def _qsize(self):
return len(self.queue)
Expand Down
6 changes: 2 additions & 4 deletions sentry_sdk/_werkzeug.py
Expand Up @@ -47,8 +47,7 @@
# We need this function because Django does not give us a "pure" http header
# dict. So we might as well use it for all WSGI integrations.
#
def _get_headers(environ):
# type: (Dict[str, str]) -> Iterator[Tuple[str, str]]
def _get_headers(environ: Dict[str, str]) -> Iterator[Tuple[str, str]]:
"""
Returns only proper HTTP headers.
"""
Expand All @@ -67,8 +66,7 @@ def _get_headers(environ):
# `get_host` comes from `werkzeug.wsgi.get_host`
# https://github.com/pallets/werkzeug/blob/1.0.1/src/werkzeug/wsgi.py#L145
#
def get_host(environ, use_x_forwarded_for=False):
# type: (Dict[str, str], bool) -> str
def get_host(environ: Dict[str, str], use_x_forwarded_for: bool = False) -> str:
"""
Return the host for the given WSGI environment.
"""
Expand Down
132 changes: 53 additions & 79 deletions sentry_sdk/api.py
Expand Up @@ -30,8 +30,7 @@
F = TypeVar("F", bound=Callable[..., Any])
else:

def overload(x):
# type: (T) -> T
def overload(x: T) -> T:
return x


Expand Down Expand Up @@ -60,17 +59,15 @@ def overload(x):
]


def hubmethod(f):
# type: (F) -> F
def hubmethod(f: F) -> F:
f.__doc__ = "%s\n\n%s" % (
"Alias for :py:meth:`sentry_sdk.Hub.%s`" % f.__name__,
inspect.getdoc(getattr(Hub, f.__name__)),
)
return f


def scopemethod(f):
# type: (F) -> F
def scopemethod(f: F) -> F:
f.__doc__ = "%s\n\n%s" % (
"Alias for :py:meth:`sentry_sdk.Scope.%s`" % f.__name__,
inspect.getdoc(getattr(Scope, f.__name__)),
Expand All @@ -80,186 +77,163 @@ def scopemethod(f):

@hubmethod
def capture_event(
event, # type: Event
hint=None, # type: Optional[Hint]
scope=None, # type: Optional[Any]
**scope_kwargs # type: Any
):
# type: (...) -> Optional[str]
event: Event,
hint: Optional[Hint] = None,
scope: Optional[Any] = None,
**scope_kwargs: Any
) -> Optional[str]:
return Hub.current.capture_event(event, hint, scope=scope, **scope_kwargs)


@hubmethod
def capture_message(
message, # type: str
level=None, # type: Optional[str]
scope=None, # type: Optional[Any]
**scope_kwargs # type: Any
):
# type: (...) -> Optional[str]
message: str,
level: Optional[str] = None,
scope: Optional[Any] = None,
**scope_kwargs: Any
) -> Optional[str]:
return Hub.current.capture_message(message, level, scope=scope, **scope_kwargs)


@hubmethod
def capture_exception(
error=None, # type: Optional[Union[BaseException, ExcInfo]]
scope=None, # type: Optional[Any]
**scope_kwargs # type: Any
):
# type: (...) -> Optional[str]
error: Optional[Union[BaseException, ExcInfo]] = None,
scope: Optional[Any] = None,
**scope_kwargs: Any
) -> Optional[str]:
return Hub.current.capture_exception(error, scope=scope, **scope_kwargs)


@hubmethod
def add_breadcrumb(
crumb=None, # type: Optional[Breadcrumb]
hint=None, # type: Optional[BreadcrumbHint]
**kwargs # type: Any
):
# type: (...) -> None
crumb: Optional[Breadcrumb] = None,
hint: Optional[BreadcrumbHint] = None,
**kwargs: Any
) -> None:
return Hub.current.add_breadcrumb(crumb, hint, **kwargs)


@overload
def configure_scope():
# type: () -> ContextManager[Scope]
def configure_scope() -> ContextManager[Scope]:
pass


@overload
def configure_scope( # noqa: F811
callback, # type: Callable[[Scope], None]
):
# type: (...) -> None
callback: Callable[[Scope], None],
) -> None:
pass


@hubmethod
def configure_scope( # noqa: F811
callback=None, # type: Optional[Callable[[Scope], None]]
):
# type: (...) -> Optional[ContextManager[Scope]]
callback: Optional[Callable[[Scope], None]] = None,
) -> Optional[ContextManager[Scope]]:
return Hub.current.configure_scope(callback)


@overload
def push_scope():
# type: () -> ContextManager[Scope]
def push_scope() -> ContextManager[Scope]:
pass


@overload
def push_scope( # noqa: F811
callback, # type: Callable[[Scope], None]
):
# type: (...) -> None
callback: Callable[[Scope], None],
) -> None:
pass


@hubmethod
def push_scope( # noqa: F811
callback=None, # type: Optional[Callable[[Scope], None]]
):
# type: (...) -> Optional[ContextManager[Scope]]
callback: Optional[Callable[[Scope], None]] = None,
) -> Optional[ContextManager[Scope]]:
return Hub.current.push_scope(callback)


@scopemethod
def set_tag(key, value):
# type: (str, Any) -> None
def set_tag(key: str, value: Any) -> None:
return Hub.current.scope.set_tag(key, value)


@scopemethod
def set_context(key, value):
# type: (str, Dict[str, Any]) -> None
def set_context(key: str, value: Dict[str, Any]) -> None:
return Hub.current.scope.set_context(key, value)


@scopemethod
def set_extra(key, value):
# type: (str, Any) -> None
def set_extra(key: str, value: Any) -> None:
return Hub.current.scope.set_extra(key, value)


@scopemethod
def set_user(value):
# type: (Optional[Dict[str, Any]]) -> None
def set_user(value: Optional[Dict[str, Any]]) -> None:
return Hub.current.scope.set_user(value)


@scopemethod
def set_level(value):
# type: (str) -> None
def set_level(value: str) -> None:
return Hub.current.scope.set_level(value)


@hubmethod
def flush(
timeout=None, # type: Optional[float]
callback=None, # type: Optional[Callable[[int, float], None]]
):
# type: (...) -> None
timeout: Optional[float] = None,
callback: Optional[Callable[[int, float], None]] = None,
) -> None:
return Hub.current.flush(timeout=timeout, callback=callback)


@hubmethod
def last_event_id():
# type: () -> Optional[str]
def last_event_id() -> Optional[str]:
return Hub.current.last_event_id()


@hubmethod
def start_span(
span=None, # type: Optional[Span]
**kwargs # type: Any
):
# type: (...) -> Span
def start_span(span: Optional[Span] = None, **kwargs: Any) -> Span:
return Hub.current.start_span(span=span, **kwargs)


@hubmethod
def start_transaction(
transaction=None, # type: Optional[Transaction]
**kwargs # type: Any
):
# type: (...) -> Union[Transaction, NoOpSpan]
transaction: Optional[Transaction] = None, **kwargs: Any
) -> Union[Transaction, NoOpSpan]:
return Hub.current.start_transaction(transaction, **kwargs)


def set_measurement(name, value, unit=""):
# type: (str, float, MeasurementUnit) -> None
def set_measurement(name: str, value: float, unit: MeasurementUnit = "") -> None:
transaction = Hub.current.scope.transaction
if transaction is not None:
transaction.set_measurement(name, value, unit)


def get_current_span(hub=None):
# type: (Optional[Hub]) -> Optional[Span]
def get_current_span(hub: Optional[Hub] = None) -> Optional[Span]:
"""
Returns the currently active span if there is one running, otherwise `None`
"""
return tracing_utils.get_current_span(hub)


def get_traceparent():
# type: () -> Optional[str]
def get_traceparent() -> Optional[str]:
"""
Returns the traceparent either from the active span or from the scope.
"""
return Hub.current.get_traceparent()


def get_baggage():
# type: () -> Optional[str]
def get_baggage() -> Optional[str]:
"""
Returns Baggage either from the active span or from the scope.
"""
return Hub.current.get_baggage()


def continue_trace(environ_or_headers, op=None, name=None, source=None):
# type: (Dict[str, Any], Optional[str], Optional[str], Optional[str]) -> Transaction
def continue_trace(
environ_or_headers: Dict[str, Any],
op: Optional[str] = None,
name: Optional[str] = None,
source: Optional[str] = None,
) -> Transaction:
"""
Sets the propagation context from environment or headers and returns a transaction.
"""
Expand Down
21 changes: 9 additions & 12 deletions sentry_sdk/attachments.py
Expand Up @@ -11,13 +11,12 @@
class Attachment:
def __init__(
self,
bytes=None, # type: Union[None, bytes, Callable[[], bytes]]
filename=None, # type: Optional[str]
path=None, # type: Optional[str]
content_type=None, # type: Optional[str]
add_to_transactions=False, # type: bool
):
# type: (...) -> None
bytes: Union[None, bytes, Callable[[], bytes]] = None,
filename: Optional[str] = None,
path: Optional[str] = None,
content_type: Optional[str] = None,
add_to_transactions: bool = False,
) -> None:
if bytes is None and path is None:
raise TypeError("path or raw bytes required for attachment")
if filename is None and path is not None:
Expand All @@ -32,10 +31,9 @@ def __init__(
self.content_type = content_type
self.add_to_transactions = add_to_transactions

def to_envelope_item(self):
# type: () -> Item
def to_envelope_item(self) -> Item:
"""Returns an envelope item for this attachment."""
payload = None # type: Union[None, PayloadRef, bytes]
payload: Union[None, PayloadRef, bytes] = None
if self.bytes is not None:
if callable(self.bytes):
payload = self.bytes()
Expand All @@ -50,6 +48,5 @@ def to_envelope_item(self):
filename=self.filename,
)

def __repr__(self):
# type: () -> str
def __repr__(self) -> str:
return "<Attachment %r>" % (self.filename,)

0 comments on commit b626b17

Please sign in to comment.