Skip to content

Exceptions

Raise these from your handlers. fastgrpc translates them into the corresponding grpc.StatusCode on the wire.

from fastgrpc.exceptions import NotFound, PermissionDenied

@service
class UserService:
    @rpc
    async def get_user(self, user_id: int) -> User:
        user = await db.find(user_id)
        if user is None:
            raise NotFound(f"user {user_id} not found")
        if not has_permission(context, user_id):
            raise PermissionDenied("cannot access this user")
        return user

Mapping

Exception gRPC StatusCode
NotFound NOT_FOUND
PermissionDenied PERMISSION_DENIED
InvalidArgument INVALID_ARGUMENT
Unauthenticated UNAUTHENTICATED
AlreadyExists ALREADY_EXISTS
Internal INTERNAL
Unavailable UNAVAILABLE
DeadlineExceeded DEADLINE_EXCEEDED

Exceptions that map to gRPC status codes.

Raise these from your handlers — fastgrpc translates them into the appropriate grpc.StatusCode on the wire.

FastGrpcError

Bases: Exception

Source code in src/fastgrpc/exceptions.py 
12
13
14
15
16
17
class FastGrpcError(Exception):
    status_code: grpc.StatusCode = grpc.StatusCode.UNKNOWN

    def __init__(self, message: str = "") -> None:
        super().__init__(message)
        self.message = message

NotFound

Bases: FastGrpcError

Source code in src/fastgrpc/exceptions.py 
20
21
class NotFound(FastGrpcError):
    status_code = grpc.StatusCode.NOT_FOUND

PermissionDenied

Bases: FastGrpcError

Source code in src/fastgrpc/exceptions.py 
24
25
class PermissionDenied(FastGrpcError):
    status_code = grpc.StatusCode.PERMISSION_DENIED

InvalidArgument

Bases: FastGrpcError

Source code in src/fastgrpc/exceptions.py 
28
29
class InvalidArgument(FastGrpcError):
    status_code = grpc.StatusCode.INVALID_ARGUMENT

Unauthenticated

Bases: FastGrpcError

Source code in src/fastgrpc/exceptions.py 
32
33
class Unauthenticated(FastGrpcError):
    status_code = grpc.StatusCode.UNAUTHENTICATED

AlreadyExists

Bases: FastGrpcError

Source code in src/fastgrpc/exceptions.py 
36
37
class AlreadyExists(FastGrpcError):
    status_code = grpc.StatusCode.ALREADY_EXISTS

Internal

Bases: FastGrpcError

Source code in src/fastgrpc/exceptions.py 
40
41
class Internal(FastGrpcError):
    status_code = grpc.StatusCode.INTERNAL

Unavailable

Bases: FastGrpcError

Source code in src/fastgrpc/exceptions.py 
44
45
class Unavailable(FastGrpcError):
    status_code = grpc.StatusCode.UNAVAILABLE

DeadlineExceeded

Bases: FastGrpcError

Source code in src/fastgrpc/exceptions.py 
48
49
class DeadlineExceeded(FastGrpcError):
    status_code = grpc.StatusCode.DEADLINE_EXCEEDED