Skip to content

App

fastgrpc.app.App dataclass

Source code in src/fastgrpc/app.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
@dataclass
class App:
    host: str = "127.0.0.1"
    port: int = 50051
    max_workers: int = 10
    reflection: bool = True
    proto_dir: str | None = None
    """Directory where the human-readable .proto file is written. Visible,
    committable, refreshed on every reload. If None (default), it's emitted
    next to the entry-point .py file — your service owns its own contract."""
    proto_out: str = ".fastgrpc"
    """Directory for compiled stubs (_pb2.py / _pb2_grpc.py). Gitignored."""
    lock_path: str = ".fastgrpc.lock"
    interceptors: list[Interceptor] = field(default_factory=list)

    def interceptor(self, fn: Interceptor) -> Interceptor:
        """Register an interceptor. Usable as a decorator."""
        self.interceptors.append(fn)
        return fn

proto_dir class-attribute instance-attribute 

proto_dir: str | None = None

Directory where the human-readable .proto file is written. Visible, committable, refreshed on every reload. If None (default), it's emitted next to the entry-point .py file — your service owns its own contract.

proto_out class-attribute instance-attribute 

proto_out: str = '.fastgrpc'

Directory for compiled stubs (_pb2.py / _pb2_grpc.py). Gitignored.

interceptor 

interceptor(fn: Interceptor) -> Interceptor

Register an interceptor. Usable as a decorator.

Source code in src/fastgrpc/app.py 
34
35
36
37
def interceptor(self, fn: Interceptor) -> Interceptor:
    """Register an interceptor. Usable as a decorator."""
    self.interceptors.append(fn)
    return fn

fastgrpc.app.Interceptor

Bases: Protocol

Source code in src/fastgrpc/app.py 
 9
10
11
12
13
14
15
16
class Interceptor(Protocol):
    async def __call__(
        self,
        method: str,
        request: object,
        context: object,
        call_next: Callable[[str, object, object], Awaitable[object]],
    ) -> object: ...

Example

from fastgrpc import App, rpc, service

cfg = App(host="0.0.0.0", port=50051, max_workers=20)

@cfg.interceptor
async def logging_interceptor(method, request, context, call_next):
    print(f"→ {method}")
    response = await call_next(method, request, context)
    print(f"← {method}")
    return response

@service
class UserService:
    @rpc
    async def get_user(self, user_id: int) -> User: ...

Configuration fields

Field Default Purpose
host 127.0.0.1 Bind address
port 50051 Bind port
max_workers 10 Thread pool size (sync server only)
reflection True Enable gRPC server reflection
proto_dir None Where to write the human-readable .proto. CLI sets this to .fastgrpc/ automatically.
proto_out .fastgrpc Where compiled _pb2.py stubs go
lock_path .fastgrpc.lock Field-number lock file location
interceptors [] List of registered interceptors