opensource/opentelemetry-python-contrib
1
0
Fork 0
mirror of https://github.com/open-telemetry/opentelemetry-python-contrib.git synced 2025-07-30 21:56:07 +08:00
Code Issues Packages Projects Releases Wiki Activity

celery: remove deprecated import of ContextManager from typing (#3145)

Browse Source 
* celery: remove deprecated import of ContextManager from typing

Signed-off-by: emdneto <9735060+emdneto@users.noreply.github.com>

* add type checking

Signed-off-by: emdneto <9735060+emdneto@users.noreply.github.com>

* ruff

Signed-off-by: emdneto <9735060+emdneto@users.noreply.github.com>

---------

Signed-off-by: emdneto <9735060+emdneto@users.noreply.github.com>
This commit is contained in:
Emídio Neto
2024-12-24 10:55:33 -03:00
committed by GitHub
parent 375eb6f2cc
commit 396aad9c24
2 changed files with 9 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
instrumentation/opentelemetry-instrumentation-celery/src/opentelemetry/instrumentation/celery/utils.py
View File

@ -12,8 +12,10 @@
# See the License for the specific language governing permissions and
# limitations under the License.
from __future__ import annotations
import logging
from typing import ContextManager, Optional, Tuple
from typing import TYPE_CHECKING, Optional, Tuple
from celery import registry # pylint: disable=no-name-in-module
from celery.app.task import Task
@ -21,6 +23,9 @@ from celery.app.task import Task
from opentelemetry.semconv.trace import SpanAttributes
from opentelemetry.trace import Span
if TYPE_CHECKING:
from contextlib import AbstractContextManager
logger = logging.getLogger(__name__)
# Celery Context key
@ -123,7 +128,7 @@ def attach_context(
task: Optional[Task],
task_id: str,
span: Span,
activation: ContextManager[Span],
activation: AbstractContextManager[Span],
token: Optional[object],
is_publish: bool = False,
) -> None:
@ -171,7 +176,7 @@ def detach_context(task, task_id, is_publish=False) -> None:
def retrieve_context(
task, task_id, is_publish=False
) -> Optional[Tuple[Span, ContextManager[Span], Optional[object]]]:
) -> Optional[Tuple[Span, AbstractContextManager[Span], Optional[object]]]:
"""Helper to retrieve an active `Span`, `ContextManager` and context token
stored in a `Task` instance
"""

2
instrumentation/opentelemetry-instrumentation-celery/test-requirements-1.txt
View File

@ -1,6 +1,6 @@
amqp==5.2.0
asgiref==3.8.1
billiard==4.2.0
billiard==4.2.1
celery==5.3.6
click==8.1.7
click-didyoumean==0.3.0