/
template.go
106 lines (89 loc) · 2.54 KB
/
template.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
package generator
import "text/template"
type ConnecpyTemplateVariables struct {
FileName string
ModuleName string
Services []*ConnecpyService
}
type ConnecpyService struct {
ServiceURL string
Name string
Comment string
Methods []*ConnecpyMethod
}
type ConnecpyMethod struct {
ServiceURL string
ServiceName string
Name string
Comment string
InputType string
OutputType string
}
type ConnecpyImport struct {
From string
Import string
}
// ConnecpyTemplate - Template for connecpy server and client
var ConnecpyTemplate = template.Must(template.New("ConnecpyTemplate").Parse(`# -*- coding: utf-8 -*-
# Generated by https://github.com/i2y/connecpy/protoc-gen-connecpy. DO NOT EDIT!
# source: {{.FileName}}
from typing import Protocol, Union
import httpx
from connecpy.async_client import AsyncConnecpyClient
from connecpy.base import Endpoint
from connecpy.server import ConnecpyServer
from connecpy.client import ConnecpyClient
from connecpy.context import ServiceContext
import {{.ModuleName}}_pb2 as _pb2
{{range .Services}}
class {{.Name}}Service(Protocol):{{- range .Methods }}
async def {{.Name}}(self, req: _pb2.{{.InputType}}, ctx: ServiceContext) -> _pb2.{{.OutputType}}:
...{{- end }}
class {{.Name}}Server(ConnecpyServer):
def __init__(self, *, service: {{.Name}}Service, server_path_prefix=""):
super().__init__(service=service)
self._prefix = f"{server_path_prefix}/{{.ServiceURL}}"
self._endpoints = { {{- range .Methods }}
"{{.Name}}": Endpoint[_pb2.{{.InputType}}, _pb2.{{.OutputType}}](
service_name="{{.ServiceName}}",
name="{{.Name}}",
function=getattr(service, "{{.Name}}"),
input=_pb2.{{.InputType}},
output=_pb2.{{.OutputType}},
),{{- end }}
}
class {{.Name}}Client(ConnecpyClient):{{range .Methods}}
def {{.Name}}(
self,
*,
request,
ctx,
server_path_prefix="",
**kwargs,
):
return self._make_request(
url=f"{server_path_prefix}/{{.ServiceURL}}/{{.Name}}",
ctx=ctx,
request=request,
response_obj=_pb2.{{.OutputType}},
**kwargs,
){{end}}
class Async{{.Name}}Client(AsyncConnecpyClient):{{range .Methods}}
async def {{.Name}}(
self,
*,
request,
ctx,
server_path_prefix="",
session: Union[httpx.AsyncClient, None] = None,
**kwargs,
):
return await self._make_request(
url=f"{server_path_prefix}/{{.ServiceURL}}/{{.Name}}",
ctx=ctx,
request=request,
response_obj=_pb2.{{.OutputType}},
session=session,
**kwargs,
)
{{end}}{{end}}`))