thrift/compiler/generate/templates/python/lite_services.py.mustache

{{!

  Copyright (c) Meta Platforms, Inc. and affiliates.

  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

      http://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.

}}{{!
Generates a top-level file to be imported in the user's service code. It provides
wrappers for each of the service handlers that the user is then able to extend.
}}
{{> common/auto_generated_py}}
from abc import ABCMeta
import typing as _typing

import folly.iobuf as _fbthrift_iobuf

import {{program:root_module_prefix}}apache.thrift.metadata.lite_types as _fbthrift_metadata
from {{program:base_library_package}}.serializer import serialize_iobuf, deserialize, Protocol
from {{program:base_library_package}}.server import ServiceInterface, oneway, PythonUserException

import {{program:module_path}}.lite_types
import {{program:module_path}}.lite_metadata
{{#program:include_namespaces}}
{{#has_services?}}
import {{included_module_path}}.lite_services
{{/has_services?}}
{{#has_types?}}
import {{included_module_path}}.lite_types
{{/has_types?}}
{{/program:include_namespaces}}

{{#program:services}}
class {{service:name}}Interface(
  {{#service:extends}}{{#service:external_program?}}
    {{service:module_path}}.lite_services.{{/service:external_program?}}{{service:name}}Interface,
    {{/service:extends}}
    {{^service:extends?}}
    ServiceInterface,
    {{/service:extends?}}
    metaclass=ABCMeta
):

    @staticmethod
    def service_name() -> bytes:
        return b"{{service:name}}"

    # pyre-ignore[3]: it can return anything
    def getFunctionTable(self) -> _typing.Mapping[bytes, _typing.Callable[..., _typing.Any]]:
        functionTable = {
        {{#service:supported_functions}}
            b"{{function:name}}": self._fbthrift__handler_{{function:name}},
        {{/service:supported_functions}}
        }
        return {**super().getFunctionTable(), **functionTable}

    @staticmethod
    def __get_thrift_name__() -> str:
        return "{{program:name}}.{{service:name}}"

    @staticmethod
    def __get_metadata__() -> _fbthrift_metadata.ThriftMetadata:
        return {{program:module_path}}.lite_metadata.gen_metadata_service_{{service:name}}()


{{#service:supported_functions}}

    async def {{function:name}}(
            self{{#function:args}},
            {{field:py_name}}: {{#field:type}}{{> types/pep484_type}}{{/field:type}}{{/function:args}}
        ) -> {{#function:return_type}}{{> types/pep484_type}}{{/function:return_type}}:
        raise NotImplementedError("async def {{function:name}} is not implemented")

    {{#function:oneway?}}
    @oneway
    {{/function:oneway?}}
    async def _fbthrift__handler_{{function:name}}(self, args: _fbthrift_iobuf.IOBuf, protocol: Protocol) {{!
            }}-> {{#function:oneway?}}None{{/function:oneway?}}{{^function:oneway?}}_fbthrift_iobuf.IOBuf{{/function:oneway?}}:
        args_struct = deserialize({{> types/function_args_type}}, args, protocol)
        {{#function:exceptions?}}
        try:
            value = await self.{{function:name}}({{#function:args}}args_struct.{{field:py_name}},{{/function:args}})
            {{^function:oneway?}}
            return_struct = {{> types/function_return_type}}({{!
            }}{{#function:return_type}}{{^type:void?}}success=value{{/type:void?}}{{/function:return_type}})
            {{/function:oneway?}}
        {{#function:exceptions}}{{#field:type}}
        except {{type:module_path}}.{{#type:struct}}{{struct:name}}{{/type:struct}} as e:
            return_struct = {{> types/function_return_type}}({{field:py_name}}=e)
            buf = serialize_iobuf(return_struct, protocol)
            exp = PythonUserException('{{type:py3_namespace}}{{#type:struct}}{{struct:name}}{{/type:struct}}', str(e), buf)
            raise exp
        {{/field:type}}
        {{/function:exceptions}}
        {{/function:exceptions?}}
        {{^function:exceptions?}}
        value = await self.{{function:name}}({{#function:args}}args_struct.{{field:py_name}},{{/function:args}})
        {{^function:oneway?}}
        return_struct = {{> types/function_return_type}}({{!
        }}{{#function:return_type}}{{^type:void?}}success=value{{/type:void?}}{{/function:return_type}})
        {{/function:oneway?}}
        {{/function:exceptions?}}

        {{^function:oneway?}}
        return serialize_iobuf(return_struct, protocol)
        {{/function:oneway?}}

{{/service:supported_functions}}
{{/program:services}}