uvloop fixes (#2297)

- Updates required for uvloop >= 0.22.1, following deprecations in the python API
regarding async loop policies.

- Updates to pytest to run tests using async correctly

- Updates to runtimes so that pytest doesn't load the same module (conftest.py)
multiple times. This involves adding __init__.py files in intermediate
directories to explicitly define python packages.

Author: lc525

mlserver/batching/requests.py

@@ -18,6 +18,7 @@ def _get_data(payload: Union[RequestInput, ResponseOutput]):
 
 def _get_parameters(payload: ResponseOutput) -> DefaultDict[Any, Any]:
     parameters = defaultdict(list)
+    payload_parameters = {}
     if payload.parameters is not None:
         payload_parameters = payload.parameters.model_dump()
     for param_name, param_values in payload_parameters.items():

mlserver/cli/main.py

@@ -3,36 +3,40 @@
 """
 
 import click
-import asyncio
 
 from functools import wraps
 
 from .init_project import init_cookiecutter_project
 
 from ..server import MLServer
 from ..logging import logger, configure_logger
-from ..utils import install_uvloop_event_loop
+from ..utils import AsyncManager
 
 from .build import generate_dockerfile, build_image, write_dockerfile
 from .serve import load_settings
 from ..batch_processing import process_batch, CHOICES_TRANSPORT
 
+CTX_ASYNC_MGR_KEY = "async_manager"
+
 
 def click_async(f):
     @wraps(f)
     def wrapper(*args, **kwargs):
-        return asyncio.run(f(*args, **kwargs))
+        ctx = click.get_current_context()
+        async_mgr = ctx.obj[CTX_ASYNC_MGR_KEY]
+        return async_mgr.run(f(*args, **kwargs))
 
     return wrapper
 
 
 @click.group()
 @click.version_option()
-def root():
+@click.pass_context
+def root(ctx):
     """
     Command-line interface to manage MLServer models.
     """
-    pass
+    ctx.ensure_object(dict)
 
 
 @root.command("start")
@@ -265,8 +269,8 @@ async def infer(
 
 def main():
     configure_logger()
-    install_uvloop_event_loop()
-    root()
+    async_mgr = AsyncManager()
+    root(obj={CTX_ASYNC_MGR_KEY: async_mgr})
 
 
 if __name__ == "__main__":

mlserver/codecs/__init__.py

@@ -1,5 +1,6 @@
+from typing import TYPE_CHECKING
+
 from .numpy import NumpyCodec, NumpyRequestCodec
-from .pandas import PandasCodec
 from .string import StringCodec, StringRequestCodec
 from .base64 import Base64Codec
 from .datetime import DatetimeCodec
@@ -24,6 +25,9 @@
     decode_inference_request,
 )
 
+if TYPE_CHECKING:  # pragma: no cover - type checking only
+    from .pandas import PandasCodec  # noqa: F401
+
 __all__ = [
     "CodecError",
     "NumpyCodec",
@@ -49,3 +53,22 @@
     "decode_inference_request",
     "decode_args",
 ]
+
+
+def __getattr__(name: str):  # pragma: no cover - lightweight lazy import
+    if name == "PandasCodec":
+        return _load_pandas_codec()
+
+    raise AttributeError(f"module 'mlserver.codecs' has no attribute {name!r}")
+
+
+def _load_pandas_codec():
+    try:
+        from .pandas import PandasCodec as _PandasCodec  # Local import to stay optional
+    except Exception as exc:  # pragma: no cover - propagate useful context
+        raise ImportError(
+            "PandasCodec requires the optional 'pandas' dependency"
+        ) from exc
+
+    globals()["PandasCodec"] = _PandasCodec
+    return _PandasCodec

mlserver/codecs/pandas.py

@@ -115,7 +115,7 @@ def encode_response(
         payload: pd.DataFrame,
         model_version: Optional[str] = None,
         use_bytes: bool = True,
-        **kwargs
+        **kwargs,
     ) -> InferenceResponse:
         outputs = cls.encode_outputs(payload, use_bytes=use_bytes)