# Source code for opentelemetry._logs._internal

    # Copyright The OpenTelemetry Authors
    # SPDX-License-Identifier: Apache-2.0
    """
    The OpenTelemetry logging API describes the classes used to generate logs and events.

    The :class:`.LoggerProvider` provides users access to the :class:`.Logger`.

    This module provides abstract (i.e. unimplemented) classes required for
    logging, and a concrete no-op implementation :class:`.NoOpLogger` that allows applications
    to use the API package alone without a supporting implementation.

    To get a logger, you need to provide the package name from which you are
    calling the logging APIs to OpenTelemetry by calling `LoggerProvider.get_logger`
    with the calling module name and the version of your package.

    The following code shows how to obtain a logger using the global :class:`.LoggerProvider`::

        from opentelemetry._logs import get_logger

        logger = get_logger("example-logger")

    .. versionadded:: 1.15.0
    """

    from __future__ import annotations

    from abc import ABC, abstractmethod
    from logging import getLogger
    from os import environ
    from time import time_ns
    from typing import cast, overload

    from typing_extensions import deprecated

    from opentelemetry._logs.severity import SeverityNumber
    from opentelemetry.context import get_current
    from opentelemetry.context.context import Context
    from opentelemetry.environment_variables import _OTEL_PYTHON_LOGGER_PROVIDER
    from opentelemetry.trace import get_current_span
    from opentelemetry.trace.span import TraceFlags
    from opentelemetry.util._once import Once
    from opentelemetry.util._providers import _load_provider
    from opentelemetry.util.types import AnyValue, _ExtendedAttributes

    _logger = getLogger(__name__)

    class LogRecord(ABC):
        """A LogRecord instance represents an event being logged.

        LogRecord instances are created and emitted via `Logger`
        every time something is logged. They contain all the information
        pertinent to the event being logged.
        """

        @overload
        def __init__(
            self,
            *,
            timestamp: int | None = None,
            observed_timestamp: int | None = None,
            context: Context | None = None,
            severity_text: str | None = None,
            severity_number: SeverityNumber | None = None,
            body: AnyValue = None,
            attributes: _ExtendedAttributes | None = None,
            event_name: str | None = None,
            exception: BaseException | None = None,
        ) -> None: ...

        @overload
        @deprecated(
            "LogRecord init with `trace_id`, `span_id`, and/or `trace_flags` is deprecated since 1.35.0. Use `context` instead."
        )
        def __init__(
            self,
            *,
            timestamp: int | None = None,
            observed_timestamp: int | None = None,
            trace_id: int | None = None,
            span_id: int | None = None,
            trace_flags: TraceFlags | None = None,
            severity_text: str | None = None,
            severity_number: SeverityNumber | None = None,
            body: AnyValue = None,
            attributes: _ExtendedAttributes | None = None,
        ) -> None: ...

        def __init__(
            self,
            *,
            timestamp: int | None = None,
            observed_timestamp: int | None = None,
            context: Context | None = None,
            trace_id: int | None = None,
            span_id: int | None = None,
            trace_flags: TraceFlags | None = None,
            severity_text: str | None = None,
            severity_number: SeverityNumber | None = None,
            body: AnyValue = None,
            attributes: _ExtendedAttributes | None = None,
            event_name: str | None = None,
            exception: BaseException | None = None,
        ) -> None:
            if not context:
                context = get_current()
            span_context = get_current_span(context).get_span_context()
            self.timestamp = timestamp
            if observed_timestamp is None:
                observed_timestamp = time_ns()
            self.observed_timestamp = observed_timestamp
            self.context = context
            self.trace_id = trace_id or span_context.trace_id
            self.span_id = span_id or span_context.span_id
            self.trace_flags = trace_flags or span_context.trace_flags
            self.severity_text = severity_text
            self.severity_number = severity_number
            self.body = body
            self.attributes = attributes
            self.event_name = event_name
            self.exception = exception

    class Logger(ABC):
        """Handles emitting events and logs via `LogRecord`."""

        def __init__(
            self,
            name: str,
            version: str | None = None,
            schema_url: str | None = None,
            attributes: _ExtendedAttributes | None = None,
        ) -> None:
            super().__init__()
            self._name = name
            self._version = version
            self._schema_url = schema_url
            self._attributes = attributes

        @overload
        def emit(
            self,
            *,
            timestamp: int | None = None,
            observed_timestamp: int | None = None,
            context: Context | None = None,
            severity_number: SeverityNumber | None = None,
            severity_text: str | None = None,
            body: AnyValue | None = None,
            attributes: _ExtendedAttributes | None = None,
            event_name: str | None = None,
            exception: BaseException | None = None,
        ) -> None: ...

        @overload
        def emit(
            self,
            record: LogRecord,
        ) -> None: ...

        @abstractmethod
        def emit(
            self,
            record: LogRecord | None = None,
            *,
            timestamp: int | None = None,
            observed_timestamp: int | None = None,
            context: Context | None = None,
            severity_number: SeverityNumber | None = None,
            severity_text: str | None = None,
            body: AnyValue | None = None,
            attributes: _ExtendedAttributes | None = None,
            event_name: str | None = None,
            exception: BaseException | None = None,
        ) -> None:
            """Emits a :class:`LogRecord` representing a log to the processing pipeline."""

    class NoOpLogger(Logger):
        """The default Logger used when no Logger implementation is available.

        All operations are no-op.
        """

        @overload
        def emit(
            self,
            *,
            timestamp: int | None = None,
            observed_timestamp: int | None = None,
            context: Context | None = None,
            severity_number: SeverityNumber | None = None,
            severity_text: str | None = None,
            body: AnyValue | None = None,
            attributes: _ExtendedAttributes | None = None,
            event_name: str | None = None,
            exception: BaseException | None = None,
        ) -> None: ...

        @overload
        def emit(  # pylint:disable=arguments-differ
            self,
            record: LogRecord,
        ) -> None: ...

        def emit(
            self,
            record: LogRecord | None = None,
            *,
            timestamp: int | None = None,
            observed_timestamp: int | None = None,
            context: Context | None = None,
            severity_number: SeverityNumber | None = None,
            severity_text: str | None = None,
            body: AnyValue | None = None,
            attributes: _ExtendedAttributes | None = None,
            event_name: str | None = None,
            exception: BaseException | None = None,
        ) -> None:
            pass

    class ProxyLogger(Logger):
        def __init__(  # pylint: disable=super-init-not-called
            self,
            name: str,
            version: str | None = None,
            schema_url: str | None = None,
            attributes: _ExtendedAttributes | None = None,
        ):
            self._name = name
            self._version = version
            self._schema_url = schema_url
            self._attributes = attributes
            self._real_logger: Logger | None = None
            self._noop_logger = NoOpLogger(name)

        @property
        def _logger(self) -> Logger:
            if self._real_logger:
                return self._real_logger

            if _LOGGER_PROVIDER:
                self._real_logger = _LOGGER_PROVIDER.get_logger(
                    self._name,
                    self._version,
                    self._schema_url,
                    self._attributes,
                )
                return self._real_logger
            return self._noop_logger

        @overload
        def emit(
            self,
            *,
            timestamp: int | None = None,
            observed_timestamp: int | None = None,
            context: Context | None = None,
            severity_number: SeverityNumber | None = None,
            severity_text: str | None = None,
            body: AnyValue | None = None,
            attributes: _ExtendedAttributes | None = None,
            event_name: str | None = None,
            exception: BaseException | None = None,
        ) -> None: ...

        @overload
        def emit(  # pylint:disable=arguments-differ
            self,
            record: LogRecord,
        ) -> None: ...

        def emit(
            self,
            record: LogRecord | None = None,
            *,
            timestamp: int | None = None,
            observed_timestamp: int | None = None,
            context: Context | None = None,
            severity_number: SeverityNumber | None = None,
            severity_text: str | None = None,
            body: AnyValue | None = None,
            attributes: _ExtendedAttributes | None = None,
            event_name: str | None = None,
            exception: BaseException | None = None,
        ) -> None:
            if record:
                self._logger.emit(record)
            else:
                self._logger.emit(
                    timestamp=timestamp,
                    observed_timestamp=observed_timestamp,
                    context=context,
                    severity_number=severity_number,
                    severity_text=severity_text,
                    body=body,
                    attributes=attributes,
                    event_name=event_name,
                    exception=exception,
                )

    class LoggerProvider(ABC):
        """
        LoggerProvider is the entry point of the API. It provides access to Logger instances.
        """

        @abstractmethod
        def get_logger(
            self,
            name: str,
            version: str | None = None,
            schema_url: str | None = None,
            attributes: _ExtendedAttributes | None = None,
        ) -> Logger:
            """Returns a `Logger` for use by the given instrumentation library.

            For any two calls with identical parameters, it is undefined whether the same
            or different `Logger` instances are returned.

            This function may return different `Logger` types (e.g. a no-op logger
            vs. a functional logger).

            Args:
                name: The name of the instrumenting module, package or class.
                    This should *not* be the name of the module, package or class that is
                    instrumented but the name of the code doing the instrumentation.
                    E.g., instead of ``"requests"``, use
                    ``"opentelemetry.instrumentation.requests"``.

                    For log sources which define a logger name (e.g. logging.Logger.name)
                    the Logger Name should be recorded as the instrumentation scope name.

                version: Optional. The version string of the
                    instrumenting library.  Usually this should be the same as
                    ``importlib.metadata.version(instrumenting_library_name)``.

                schema_url: Optional. Specifies the Schema URL of the emitted telemetry.

                attributes: Optional. Specifies the instrumentation scope attributes to
                    associate with emitted telemetry.
            """

    class NoOpLoggerProvider(LoggerProvider):
        """The default LoggerProvider used when no LoggerProvider implementation is available."""

        def get_logger(
            self,
            name: str,
            version: str | None = None,
            schema_url: str | None = None,
            attributes: _ExtendedAttributes | None = None,
        ) -> Logger:
            """Returns a NoOpLogger."""
            return NoOpLogger(
                name, version=version, schema_url=schema_url, attributes=attributes
            )

    class ProxyLoggerProvider(LoggerProvider):
        def get_logger(
            self,
            name: str,
            version: str | None = None,
            schema_url: str | None = None,
            attributes: _ExtendedAttributes | None = None,
        ) -> Logger:
            if _LOGGER_PROVIDER:
                return _LOGGER_PROVIDER.get_logger(
                    name,
                    version=version,
                    schema_url=schema_url,
                    attributes=attributes,
                )
            return ProxyLogger(
                name,
                version=version,
                schema_url=schema_url,
                attributes=attributes,
            )

    _LOGGER_PROVIDER_SET_ONCE = Once()
    _LOGGER_PROVIDER: LoggerProvider | None = None
    _PROXY_LOGGER_PROVIDER = ProxyLoggerProvider()

    def get_logger_provider() -> LoggerProvider:
        """Gets the current global :class:`~.LoggerProvider` object."""
        global _LOGGER_PROVIDER  # pylint: disable=global-variable-not-assigned
        if _LOGGER_PROVIDER is None:
            if _OTEL_PYTHON_LOGGER_PROVIDER not in environ:
                return _PROXY_LOGGER_PROVIDER

            logger_provider: LoggerProvider = _load_provider(  # type: ignore
                _OTEL_PYTHON_LOGGER_PROVIDER, "logger_provider"
            )
            _set_logger_provider(logger_provider, log=False)

        # _LOGGER_PROVIDER will have been set by one thread
        return cast("LoggerProvider", _LOGGER_PROVIDER)

    def _set_logger_provider(logger_provider: LoggerProvider, log: bool) -> None:
        def set_lp() -> None:
            global _LOGGER_PROVIDER  # pylint: disable=global-statement
            _LOGGER_PROVIDER = logger_provider

        did_set = _LOGGER_PROVIDER_SET_ONCE.do_once(set_lp)

        if log and not did_set:
            _logger.warning("Overriding of current LoggerProvider is not allowed")

    def set_logger_provider(logger_provider: LoggerProvider) -> None:
        """Sets the current global :class:`~.LoggerProvider` object.

        This can only be done once, a warning will be logged if any further attempt
        is made.
        """
        _set_logger_provider(logger_provider, log=True)

    def get_logger(
        instrumenting_module_name: str,
        instrumenting_library_version: str = "",
        logger_provider: LoggerProvider | None = None,
        schema_url: str | None = None,
        attributes: _ExtendedAttributes | None = None,
    ) -> Logger:
        """Returns a `Logger` for use within a python process.

        This function is a convenience wrapper for
        opentelemetry.sdk._logs.LoggerProvider.get_logger.

        If logger_provider param is omitted the current configured one is used.
        """
        if logger_provider is None:
            logger_provider = get_logger_provider()
        return logger_provider.get_logger(
            instrumenting_module_name,
            instrumenting_library_version,
            schema_url,
            attributes,
        )