Yeah I can reproduce that. I also see a similar error of I try and assign to a typed variable.

from kr8s.objects import Pod

pods: list[Pod] = Pod.list()

$ mypy test.py
test.py:3: error: Incompatible types in assignment (expression has type "Coroutine[Any, Any, APIObject | list[APIObject]]", variable has type "list[Pod]")  [assignment]
Found 1 error in 1 file (checked 1 source file)

I think there are two potential solutions here:

• Find some way to tell mypy that the @sync wrapper changes the type
  • We already do this in sphinx-autoapi via a plugin. Maybe there is a way to do this for mypy?
• Provide a .pyi stub file for each sync API submodule to override the typing
  • I would want to find a way to add tests to ensure this stays in sync. Otherwise we just introduce a new class of bug where the stubs get out of sync with the code. I had a quick play around with stubgen and stubtest but I'm not sure if they will run into similar issues as mypy around incorrectly identifying the signatures of wrapped classes.

A third option could be to do away with the @sync wrapper and manually create each method and use run_sync() directly. This is what we do for functions like kr8s.get() today.

# kr8s/objects.py
from kr8s._objects import Pod as _Pod
from kr8s._async_utils import run_sync

class Pod(_Pod):
    """..."""

    ...

    @classmethod
    def list(cls, **kwargs) -> APIObject | list[APIObject]:
        """List objects in Kubernetes.

        Args:
            **kwargs: Keyword arguments to pass to :func:`kr8s.get`.

        Returns:
            A list of objects.
        """
        return run_sync(super().list)(**kwargs)

    ...

The upsides of this are:

• Static analysers like mypy would have an easier time parsing the code
• We could do away with the sphinx-autoapi plugin

But the downsides are:

• We would need to duplicate all methods, including inherited methods from the base class
• We would need to duplicate docstrings
• This would potentially add thousands of lines of boilerplate code
• We would need to find a way to keep everything in sync

I've tried several approaches, but none seem to work for the first option.

The third option is also problematic because:

• MyPy doesn't allow overriding an async method with a sync one,
• The best return type would be list[Pod], but super().list actually returns an async APIObject.

There doesn't seem to be an easy fix

Thanks for taking the time to dig into this.

I did think more about option 1 and I guess it's problematic because even if we write a plugin specifically for kr8s, all code that depends on kr8s would need the plugin, which is not something we can ask users to do.

I agree for option 3 it would be best to return list[Pod]. We could add a runtime assertion to make mypy happy and narrow the type. But as you say if you can't override async methods with sync ones then this option is not viable.

For option 2 I wonder if we need some custom stub generation script that we can invoke as part of the pre-commit hooks. This way we can explicitly tell mypy what to expect, but we don't need to worry about drift if they are auto-generated.

Chiming in here, the easiest would be just separating the api surface

class Pod(ApiObject):
    async def get_async(self) -> <ASYNC TYPEHINT>

    def get(self) -> Pod:
          <call get_async function here with asyncio>
          

@ion-elgreco I appreciate the input by that's not quite how things work in this library. All objects are created as async first where all methods are async, and then wrapped with a decorator that makes them sync to expose the sync API.

I think what you are suggesting is option 3, which is to do away with the @sync wrapper and duplicate out the whole API a second time so that we can explicitly set type hints correctly on the sync API.

@jacobtomlinson yes that option, but probably with some clever inheritance you could reduce the duplication. It would be definitely a breaking change and warrant for a 1.0 perhaps

@ion-elgreco if you have some thoughts on how to do this I'd be really interested to hear. I've been thinking about this for months at this point and still haven't found an elegant solution that doesn't involve a whole bunch of code duplication.

The best option I've found so far is to make a script that generates type stubs for the sync API (see #493). This way we can leave the async->sync conversion as it is and just fix the type hints after the fact. But this is non-trivial and I havent got a robust implementation finished yet.

I'm keen to explore separating out the APIs though if there is a better implementation. For example:

• The base class kr8s.asyncio.objects.APIObject contains async implementations of methods like .create().
• The child class kr8s.asyncio.objects.Pod adds additional async methods like .logs().
• The child class kr8s.objects.Pod adds the @sync decorator to expose a sync API.

If we removed the @sync decorator from the sync Pod implementation we would need to stub out every method from APIObject and Pod and call the corotuine from the super in a sync way and set the type hints correctly. Doing this for every object would result in a lot of duplicate definitions of base methods like .create(), .delete(), etc. This has always put me off going down this road.

Are you suggesting there is something we could do with a mixin to remove this duplication?

@jacobtomlinson There aren't many solutions tbh. You always have to declare the sync method explicitly. Maintainability wise, this seems simpler

class APIObject(APIObjectBase):
    def get(
        cls,
        name: str | None = None,
        namespace: str | None = None,
        api: Api | None = None,
        label_selector: str | dict[str, str] | None = None,
        field_selector: str | dict[str, str] | None = None,
        timeout: int = 2,
        **kwargs,
    ) -> Self:
        return run_sync(cls.get_async(
            name=name,
            namespace=namespace,
            api=api,
            label_selector=label_selector,
            field_selector=field_selector,
            timeout=timeout,
            kwargs=kwargs
        )) # type: ignore

Or it's dynamically adding methods and creating the stubs, like you did

Sure. I guess I'm intrigured by your commend about clever inheritance stuff to reduce duplication. What did you have in mind?

Sure. I guess I'm intrigured by your commend about clever inheritance stuff to reduce duplication. What did you have in mind?

Nothing fancy, just having a private Bass class holds async methods that are reused across the api surface. Main APIObject inherits that and you explicitly set the sync versions of it, which calls the async ones. This would be more static and just once some work upfront.