Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Add stubs for grpc.aio #13

Merged
merged 1 commit into from Feb 24, 2021

Conversation

ColdrickSotK
Copy link
Contributor

The grpc.aio module has good type annotation, but without these
stubs existing mypy falls over when using it whilst the gRPC stubs
in this repository are installed.

The stubs themselves are autogenerated with stubgen and tidied up
by hand.

The grpc.aio module has good type annotation, but without these
stubs existing mypy falls over when using it whilst the gRPC stubs
in this repository are installed.

The stubs themselves are autogenerated with `stubgen` and tidied up
by hand.
@shabbyrobe
Copy link
Owner

shabbyrobe commented Feb 11, 2021

Looks good. Thanks for the PR! Would you mind providing a few representative, anonymised snippets of code where you're using this module?

@ColdrickSotK
Copy link
Contributor Author

Sorry for the delay in replying. We're currently only using the async channel constructors and client-side interceptors, and still using the synchronous code already covered for our server-side implementation.

This is pretty representative of the extent of our use currently, and reproduces the issue solved by this PR.

from grpc import aio


class ExampleInterceptor(aio.UnaryUnaryClientInterceptor):

    def __init__(self):
        self._metadata = ("example_header", "example_value")

    def _add_metadata(self, call_details):
        if call_details.metadata is not None:
            new_metadata = list(call_details.metadata)
        else:
            new_metadata = []
        
        new_metadata.append(self._metadata)
        return aio.ClientCallDetails(
            call_details.method,
            call_details.timeout,
            new_metadata,
            call_details.credentials,
            call_details.wait_for_ready
        )

    async def intercept_unary_unary(self, continuation, client_call_details, request):
        new_details = self._add_metadata(client_call_details)
        return await continuation(new_details, request)


channel = aio.insecure_channel("localhost:50051", interceptors=[ExampleInterceptor()])

# Do some gRPC requests

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Labels
None yet
Projects
None yet
Development

Successfully merging this pull request may close these issues.

None yet

2 participants