Why Gemfury? Push, build, and install  RubyGems npm packages Python packages Maven artifacts PHP packages Go Modules Debian packages RPM packages NuGet packages

Repository URL to install this package:

Details    
namara-python / rpc / exports_twirp.py
Size: Mime:
# -*- coding: utf-8 -*-
# Generated by https://github.com/verloop/twirpy/protoc-gen-twirpy.  DO NOT EDIT!
# source: exports.proto

from google.protobuf import symbol_database as _symbol_database

from twirp.base import Endpoint
from twirp.server import TwirpServer
from twirp.client import TwirpClient

_sym_db = _symbol_database.Default()

class ExportsServiceServer(TwirpServer):

	def __init__(self, *args, service):
		super().__init__(service=service)
		self._prefix = "/twirp/exports.ExportsService"
		self._endpoints = {
			"CreateExport": Endpoint(
				service_name="ExportsService",
				name="CreateExport",
				function=getattr(service, "CreateExport"),
				input=_sym_db.GetSymbol("exports.CreateExportRequest"),
				output=_sym_db.GetSymbol("exports.CreateExportResponse"),
			),
			"GetExport": Endpoint(
				service_name="ExportsService",
				name="GetExport",
				function=getattr(service, "GetExport"),
				input=_sym_db.GetSymbol("exports.GetExportRequest"),
				output=_sym_db.GetSymbol("exports.GetExportResponse"),
			),
			"ListExports": Endpoint(
				service_name="ExportsService",
				name="ListExports",
				function=getattr(service, "ListExports"),
				input=_sym_db.GetSymbol("exports.ListExportsRequest"),
				output=_sym_db.GetSymbol("exports.ListExportsResponse"),
			),
			"DeleteExport": Endpoint(
				service_name="ExportsService",
				name="DeleteExport",
				function=getattr(service, "DeleteExport"),
				input=_sym_db.GetSymbol("exports.DeleteExportRequest"),
				output=_sym_db.GetSymbol("exports.DeleteExportResponse"),
			),
			"Healthz": Endpoint(
				service_name="ExportsService",
				name="Healthz",
				function=getattr(service, "Healthz"),
				input=_sym_db.GetSymbol("exports.Empty"),
				output=_sym_db.GetSymbol("exports.Empty"),
			),
		}

class ExportsServiceClient(TwirpClient):

	def CreateExport(self, *args, ctx, request, **kwargs):
		return self._make_request(
			url="/twirp/exports.ExportsService/CreateExport",
			ctx=ctx,
			request=request,
			response_obj=_sym_db.GetSymbol("exports.CreateExportResponse"),
			**kwargs,
		)

	def GetExport(self, *args, ctx, request, **kwargs):
		return self._make_request(
			url="/twirp/exports.ExportsService/GetExport",
			ctx=ctx,
			request=request,
			response_obj=_sym_db.GetSymbol("exports.GetExportResponse"),
			**kwargs,
		)

	def ListExports(self, *args, ctx, request, **kwargs):
		return self._make_request(
			url="/twirp/exports.ExportsService/ListExports",
			ctx=ctx,
			request=request,
			response_obj=_sym_db.GetSymbol("exports.ListExportsResponse"),
			**kwargs,
		)

	def DeleteExport(self, *args, ctx, request, **kwargs):
		return self._make_request(
			url="/twirp/exports.ExportsService/DeleteExport",
			ctx=ctx,
			request=request,
			response_obj=_sym_db.GetSymbol("exports.DeleteExportResponse"),
			**kwargs,
		)

	def Healthz(self, *args, ctx, request, **kwargs):
		return self._make_request(
			url="/twirp/exports.ExportsService/Healthz",
			ctx=ctx,
			request=request,
			response_obj=_sym_db.GetSymbol("exports.Empty"),
			**kwargs,
		)