airflow_scheduler/.venv/Lib/site-packages/neo4j/_meta.py

# Copyright (c) "Neo4j"
# Neo4j Sweden AB [https://neo4j.com]
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     https://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.


from __future__ import annotations

import asyncio
import platform
import sys
import typing as t
from functools import wraps
from inspect import isclass
from warnings import warn

from ._codec.packstream import RUST_AVAILABLE


if t.TYPE_CHECKING:
    _FuncT = t.TypeVar("_FuncT", bound=t.Callable)


# Can be automatically overridden in builds
package = "neo4j"
version = "5.28.1"
deprecated_package = False


def _compute_bolt_agent() -> dict[str, str]:
    def format_version_info(version_info):
        return "{}.{}.{}-{}-{}".format(*version_info)

    language = "Python"
    if RUST_AVAILABLE:
        language += "-Rust"

    return {
        "product": f"neo4j-python/{version}",
        "platform": (
            f"{platform.system() or 'Unknown'} "
            f"{platform.release() or 'unknown'}; "
            f"{platform.machine() or 'unknown'}"
        ),
        "language": f"{language}/{format_version_info(sys.version_info)}",
        "language_details": (
            f"{platform.python_implementation()}; "
            f"{format_version_info(sys.implementation.version)} "
            f"({', '.join(platform.python_build())}) "
            f"[{platform.python_compiler()}]"
        ),
    }


BOLT_AGENT_DICT = _compute_bolt_agent()


def _compute_user_agent() -> str:
    return (
        f'{BOLT_AGENT_DICT["product"]} '
        f'{BOLT_AGENT_DICT["language"]} '
        f'({sys.platform})'
    )


USER_AGENT = _compute_user_agent()


# Undocumented but exposed.
# Other official drivers also provide means to access the default user agent.
# Hence, we'll leave this here for now.
def get_user_agent():
    """
    Obtain the driver's default user agent string.

    The user agent is sent to the server after a successful handshake.
    """
    return USER_AGENT


def _id(x):
    return x


def copy_signature(_: _FuncT) -> t.Callable[[t.Callable], _FuncT]:
    return _id


# Copy globals as function locals to make sure that they are available
# during Python shutdown when the Pool is destroyed.
def deprecation_warn(message, stack_level=1, _warn=warn):
    _warn(message, category=DeprecationWarning, stacklevel=stack_level + 1)


def deprecated(message: str) -> t.Callable[[_FuncT], _FuncT]:
    """
    Decorate deprecated functions and methods.

    ::

        @deprecated("'foo' has been deprecated in favour of 'bar'")
        def foo(x):
            pass

        @property
        @deprecated("'bar' will be internal without a replacement")
        def bar(self):
            return "bar"

        @property
        def baz(self):
            return self._baz

        @baz.setter
        @deprecated("'baz' will be read-only in the future")
        def baz(self, value):
            self._baz = value

    """
    return _make_warning_decorator(message, deprecation_warn)


# TODO: 6.0 - remove this class, replace usage with PreviewWarning
class ExperimentalWarning(Warning):
    """
    Base class for warnings about experimental features.

    .. deprecated:: 5.8
        we now use "preview" instead of "experimental":
        :class:`.PreviewWarning`.
    """


def experimental_warn(message, stack_level=1):
    warn(message, category=ExperimentalWarning, stacklevel=stack_level + 1)


def experimental(message) -> t.Callable[[_FuncT], _FuncT]:
    """
    Decorate functions and methods as experimental.

    ::

        @experimental("'foo' is an experimental function and may be "
                      "removed in a future release")
        def foo(x):
            pass

    .. deprecated:: 5.8
        we now use "preview" instead of "experimental".
    """
    return _make_warning_decorator(message, experimental_warn)


# TODO: 6.0 - consider moving this to the `warnings` module
#             and not to re-export it from the top-level package `neo4j`
class PreviewWarning(Warning):
    """
    A driver feature in preview has been used.

    It might be changed without following the deprecation policy.
    See also https://github.com/neo4j/neo4j-python-driver/wiki/preview-features
    """


def preview_warn(message, stack_level=1):
    message += (
        " It might be changed without following the deprecation policy. "
        "See also "
        "https://github.com/neo4j/neo4j-python-driver/wiki/preview-features."
    )
    warn(message, category=PreviewWarning, stacklevel=stack_level + 1)


def preview(message) -> t.Callable[[_FuncT], _FuncT]:
    """
    Decorate functions and methods as preview.

    ::

        @preview("foo is a preview.")
        def foo(x):
            pass
    """
    return _make_warning_decorator(message, preview_warn)


if t.TYPE_CHECKING:

    class _WarningFunc(t.Protocol):
        def __call__(self, message: str, stack_level: int = 1) -> None: ...
else:
    _WarningFunc = object


def _make_warning_decorator(
    message: str,
    warning_func: _WarningFunc,
) -> t.Callable[[_FuncT], _FuncT]:
    def decorator(f):
        if asyncio.iscoroutinefunction(f):

            @wraps(f)
            async def inner(*args, **kwargs):
                warning_func(message, stack_level=2)
                return await f(*args, **kwargs)

            inner._without_warning = f
            return inner
        if isclass(f):
            if hasattr(f, "__init__"):
                original_init = f.__init__

                @wraps(original_init)
                def inner(self, *args, **kwargs):
                    warning_func(message, stack_level=2)
                    return original_init(self, *args, **kwargs)

                def _without_warning(cls, *args, **kwargs):
                    obj = cls.__new__(cls, *args, **kwargs)
                    original_init(obj, *args, **kwargs)
                    return obj

                f.__init__ = inner
                f._without_warning = classmethod(_without_warning)
                return f
            raise TypeError("Cannot decorate class without __init__")
        else:

            @wraps(f)
            def inner(*args, **kwargs):
                warning_func(message, stack_level=2)
                return f(*args, **kwargs)

            inner._without_warning = f
            return inner

    return decorator


# Copy globals as function locals to make sure that they are available
# during Python shutdown when the Pool is destroyed.
def unclosed_resource_warn(obj, _warn=warn):
    cls_name = obj.__class__.__name__
    msg = f"unclosed  {cls_name}: {obj!r}."
    _warn(msg, ResourceWarning, stacklevel=2, source=obj)