Skip to content

Welcome to synclane

synclane is a framework-agnostic RPC API with a smart auto-generated TypeScript client.

License codecov Tests status Docs status PyPI Downloads Python versions

Idea

The below must be enough to define an API:

class UserParams(pydantic.BaseModel):
    uid: str

class GetUsers(AbstractProcedure):
    def call(self, in_: UserParams, context) -> List[UserDetails]:
        ...

and use an automatically generated frontend TypeScript client:

import { callGetUsers } from "./src/out";

expect(callGetUsers(userParams).$promise).resolves.toEqual(listOfUserDetails);

Benefits

Automated typescript client generation

Of course, it's possible to annotate your API, export an OpenAPI schema and generate a typescript client from it. However it will lack the below nice bits.

Browser Dates done right

Javascript doesn't have a separate date type, so it uses Date for both python's date and datetime.

Hence when you pass 2000-01-01 to a browser in New York, the browser will read it as UTC datetime and then convert it to the local timezone, so it will give you Dec 31, 1991 7:00PM, which is fine if you wanted to work with a particular moment in time, but what if you wanted to display someone's date of birth? That's why lacking date type is a problem.

synclane will see that you wanted to pass python's date to the browser and will automatically prepare it in the browser, so that Jan 1st is preserved in the case above.

Browser friendly types only

synclane raises an exception if you use types, which browser won't be able to understand.

No need to define URLs

Once you name a procedure, e.g. AddUser, you just get callAddUser function in the typescript client. You don't need to define any other identifier like API endpoint url.

Enums

If your procedure in/out types include enums, they will become available in the typescript client.

Installation

pip install synclane

pydantic is the only dependency.

Usage

  1. define procedures
  2. define RPC instance, its error handling method, register procedures and dump TypeScript client code
  3. connect RPC to an API
  4. on TypeScript side: import rpcConfig and initialize:
    • rpcConfig.url: url where RPC is listening
    • rpcConfig.initFetch (optional): function, which accepts and can mutate fetch options as needed

Example

Step 1: Define procedures

import logging
from datetime import date, datetime
from enum import Enum
from typing import Generic, List, Optional, TypeVar

from pydantic import BaseModel, ValidationError, conint

from synclane import (
    AbstractAsyncProcedure,
    AbstractAsyncRpc,
    AbstractProcedure,
    AbstractRpc,
)


logger = logging.getLogger(__name__)
logger.setLevel(logging.DEBUG)


class GetObjectParams(BaseModel):
    uid: str


class AccessLevel(Enum):
    BASIC = 1
    ADMIN = 2


class UserDetails(BaseModel):
    uid: str
    name: str
    created: datetime
    dob: date
    access_level: AccessLevel


def is_authorized(context):
    # context can be anything; let it be a request here
    if context.headers.get("x-jwt-token", "") != "secret":
        raise UnauthorizedError


class GetUser(AbstractProcedure):
    PERMISSIONS = (is_authorized,)

    def call(self, in_: GetObjectParams, context) -> UserDetails:
        return UserDetails(
            uid=in_.uid,
            name="John",
            created=datetime.fromtimestamp(0),
            dob=date(1970, 1, 1),
            access_level=AccessLevel.BASIC,
        )


# LET'S ADD FAKE PAGINATION AND MAKE IT ASYNC JUST FOR EXAMPLE
class Params(BaseModel):
    page: conint(gt=0)
    created_after: Optional[datetime] = None
    dob_after: Optional[date] = None


T = TypeVar("T")


class Paginated(BaseModel, Generic[T]):
    has_next: bool
    has_prev: bool
    data: List[T]


class GetUsers(AbstractAsyncProcedure):
    PERMISSIONS = (is_authorized,)

    async def call_async(self, in_: Params, context) -> Paginated[UserDetails]:
        return {
            "has_next": True,
            "has_prev": False,
            "data": [
                UserDetails(
                    uid="4eeb24a4-ecc1-4d9a-a43c-7263c6c60a07",
                    name="John",
                    created=in_.created_after,
                    dob=in_.dob_after,
                    access_level=AccessLevel.BASIC,
                )
            ],
        }

Step 2: Define RPC, dump TS

class UnauthorizedError(Exception):
    pass


class Rpc(AbstractAsyncRpc):  # OR AbstractRpc for sync only procedures
    def prepare_exception(self, raw_data, context, exc):
        # it can be anything, but the below tries to adhere to
        # https://www.jsonrpc.org/specification
        if isinstance(exc, ValidationError):
            return {
                "code": -32600,
                "message": "Validation error",
                "details": exc.errors(
                    include_url=False,
                    include_context=False,
                    include_input=True,
                ),
            }
        if isinstance(exc, UnauthorizedError):
            return {
                "code": -32000,
                "message": "unauthorized",
            }
        logger.exception(exc)
        return {
            "code": -1,
            "message": "Internal server error",
        }


rpc = Rpc().register(GetUsers, GetUser)


# dump TypeScript client
rpc.ts_dump("src/out.ts")

Step 3.a: Connect to Django

from django.http import HttpResponse


async def index(request):
    return HttpResponse(
        await rpc.call_async(request.body, request),
        content_type="application/json",
    )

from django.http import HttpResponse


def index(request):
    return HttpResponse(
        rpc.call(request.body, request),
        content_type="application/json",
    )

Step 3.b: Connect to FastAPI

from fastapi import FastAPI, Request, Response


app_fast_api = FastAPI()


@app_fast_api.post("/")
async def read_root(request: Request):
    return Response(
        await rpc.call_async(
            await request.body(),  # always full body as is
            request,  # anything to be passed to procedures as context
        ),
        media_type="application/json",
    )

from fastapi import FastAPI, Request, Response


app_fast_api = FastAPI()


@app_fast_api.post("/")
async def read_root(request: Request):
    return Response(
        rpc.call(
            await request.body(),  # always full body as is
            request,  # anything to be passed to procedures as context
        ),
        media_type="application/json",
    )

Step 4: Use autogenerated TS client

import {
    callGetUsers,
    callGetUser,
    AccessLevel,
    rpcConfig,
} from "../src/out";

rpcConfig.url = backend.url;

// example of adding authentication
// init is fetch options - https://developer.mozilla.org/en-US/docs/Web/API/fetch
rpcConfig.initFetch = (init: RequestInit) => {
    let headers = (init.headers = init.headers || {});
    headers["X-Jwt-Token"] = "secret";
    return init;
};
expect(
    callGetUser({ uid: "4eeb24a4-ecc1-4d9a-a43c-7263c6c60a07" }).$promise,
).resolves.toEqual({
    uid: "4eeb24a4-ecc1-4d9a-a43c-7263c6c60a07",
    name: "John",
    created: new Date(new Date(0).getTimezoneOffset() * 60000),
    dob: new Date(1970, 0, 1),
    access_level: AccessLevel.BASIC,
});