Skip to main content

The python framework, providing quick way to build API wrappers. Build painless API Requests with minimal implementation

Project description

sensei

Logo Banner


Build painless API Requests with minimal implementation

Python versions PyPi Version Coverage

The Python framework, that provides a quick way to build API wrappers. Use type hints, to build requests, with little or no implementation


Documentation: https://sensei.factorycroco.com

Source code: https://github.com/CrocoFactory/sensei

Maintain: https://www.patreon.com/user/membership?u=142083211


There are key features provided by sensei:

  • Fast: Do not write any code-making request, dedicate responsibility to the function`s interface(signature)
  • Short: Avoid code duplication.
  • Sync/Async: Implement sync and async quickly, without headaches
  • Robust: Auto validation data before and after request

Quick Overview

API Wrapper should provide these features for users:

  • Provide sync and async code versions
  • Validate data before accessing the API. A better outcome is catching exceptions, thrown due to wrong data, than getting JSON explaining the reason of the error.
  • Handle QPS (Queries per second) limits.

And as a developer, you want to avoid code duplication and make routine things faster.

To follow all these principles, you either violate DRY or have to maintain bad code architecture. Sensei is a tool to avoid these issues.

Example code:

import datetime
from typing import Annotated, Any, Self
from pydantic import EmailStr, PositiveInt, AnyHttpUrl
from sensei import Router, Query, Path, APIModel, Args, pascal_case, format_str, RateLimit
from httpx import Response

router = Router('https://reqres.in/api', rate_limit=RateLimit(5, 1))


@router.model()
class BaseModel(APIModel):
    def __finalize_json__(self, json: dict[str, Any]) -> dict[str, Any]:
        return json['data']

    def __prepare_args__(self, args: Args) -> Args:
        args.headers['X-Token'] = 'secret_token'
        return args

    def __header_case__(self, s: str) -> str:
        return pascal_case(s)


class User(BaseModel):
    email: EmailStr
    id: PositiveInt
    first_name: str
    last_name: str
    avatar: AnyHttpUrl

    @classmethod
    @router.get('/users')
    def list(
            cls,
            page: Annotated[int, Query()] = 1,
            per_page: Annotated[int, Query(le=7)] = 3
    ) -> list[Self]: # Framework knows how to handle response
        ...

    @classmethod
    @router.get('/users/{id_}')
    def get(cls, id_: Annotated[int, Path(alias='id')]) -> Self: ... # Framework knows how to handle response
    
    @router.patch('/users/{id_}', skip_finalizer=True)
    def update(
            self,
            name: str,
            job: str
    ) -> datetime.datetime:  # Framework does not know how to represent response as datetime object
        ...
    
    @update.prepare
    def _update_in(self, args: Args) -> Args:  # Get id from current object
        args.url = format_str(args.url, {'id_': self.id})
        return args
    
    @update.finalize()
    def _update_out(self, response: Response) -> datetime.datetime: # Specify hook, to handle response instead of framework
        json_ = response.json()
        result = datetime.datetime.strptime(json_['updated_at'], "%Y-%m-%dT%H:%M:%S.%fZ")
        self.first_name = json_['name']
        return result

    @router.delete('/users/{id_}')
    def delete(self) -> Self: ...
    
    @delete.prepare
    def _delete_in(self, args: Args) -> Args:
        url = args.url
        url = format_str(url, {'id_': self.id})
        args.url = url
        return args

    @router.post('/token')
    def login(self) -> str: ...

    @login.prepare
    def _login_in(self, args: Args) -> Args:
        args.json_['email'] = self.email
        return args

    @login.finalize
    def _login_out(self, response: Response) -> str:
        return response.json()['token']

    @router.put('/users/{id_}', skip_finalizer=True)
    def change(
            self,
            name: Annotated[str, Query()],
            job: Annotated[str, Query()]
    ) -> bytes:
        ...

    @change.prepare
    def _change_in(self, args: Args) -> Args:
        args.url = format_str(args.url, {'id_': self.id})
        return args

As you can see, the functions` body have no code. You only need to describe the interface(signature) of the API endpoint, that is path, method, body and params, headers, and cookies, and specify its types and return(response) type. In most cases, the framework knows how to handle a response to represent it as your return type, but when it cannot, you can use hooks to finalize the response or prepare arguments for request.

Example when the framework knows how to handle a response instead of you

@classmethod
@router.get('/users')
def list(
        cls,
        page: Annotated[int, Query()] = 1,
        per_page: Annotated[int, Query(le=7)] = 3
) -> list[Self]: # Framework knows how to handle response
    ...

Example when you need to specify hook:

@router.patch('/users/{id_}', skip_finalizer=True)
def update(
        self,
        name: str,
        job: str
) -> datetime.datetime:  # Framework does not know how to represent response as datetime object
    ...

@update.prepare
def _update_in(self, args: Args) -> Args:  # Get id from current object
    args.url = format_str(args.url, {'id_': self.id})
    return args

@update.finalize()
def _update_out(self, response: Response) -> datetime.datetime: # Specify hook, to handle response instead of framework
    json_ = response.json()
    result = datetime.datetime.strptime(json_['updated_at'], "%Y-%m-%dT%H:%M:%S.%fZ")
    self.first_name = json_['name']
    return result

Let`s try to implement almost the same functionality, without using the framework!

import httpx
import datetime
from typing import Any, Optional
from pydantic import BaseModel, EmailStr, PositiveInt, AnyHttpUrl, Field, ValidationError

TOKEN = 'secret_token'
BASE_URL =  'https://reqres.in/api' 

class User(BaseModel):
    email: EmailStr
    id: PositiveInt
    first_name: str
    last_name: str
    avatar: AnyHttpUrl

    @classmethod
    def _get_headers(cls) -> dict[str, str]:
        return {'X-Token': TOKEN}

    @classmethod
    def _handle_response(cls, response: httpx.Response) -> dict[str, Any]:
        response.raise_for_status()
        return response.json()

    @classmethod
    def _build_url(cls, endpoint: str) -> str:
        return f'{BASE_URL}{endpoint}'

    @classmethod
    def list(cls, page: Optional[int] = Field(1, ge=1), per_page: Optional[int] = Field(3, le=7)) -> list['User']:
        try:
            cls._validate_list_args(page, per_page)
        except ValidationError as e:
            raise ValueError(f"Invalid arguments: {e}")

        url = cls._build_url('/users')
        params = {'page': page, 'per_page': per_page}
        headers = cls._get_headers()

        response = httpx.get(url, params=params, headers=headers)
        data = cls._handle_response(response)['data']

        return [cls(**user_data) for user_data in data]

    @classmethod
    def _validate_list_args(cls, page: int, per_page: int) -> None:
        class ListArgs(BaseModel):
            page: int = Field(1, ge=1)
            per_page: int = Field(3, le=7)

        ListArgs(page=page, per_page=per_page)

    @classmethod
    def get(cls, id_: int) -> 'User':
        try:
            cls._validate_id(id_)
        except ValidationError as e:
            raise ValueError(f"Invalid ID: {e}")

        url = cls._build_url(f'/users/{id_}')
        headers = cls._get_headers()

        response = httpx.get(url, headers=headers)
        data = cls._handle_response(response)['data']

        return cls(**data)

    @classmethod
    def _validate_id(cls, id_: int) -> None:
        class IDModel(BaseModel):
            id_: PositiveInt

        IDModel(id_=id_)

    def update(self, name: str, job: str) -> datetime.datetime:
        try:
            self._validate_update_args(name, job)
        except ValidationError as e:
            raise ValueError(f"Invalid arguments: {e}")

        url = self._build_url(f'/users/{self.id}')
        headers = self._get_headers()
        json_data = {'name': name, 'job': job}

        response = httpx.patch(url, headers=headers, json=json_data)
        data = self._handle_response(response)

        self.first_name = data['name']
        return datetime.datetime.strptime(data['updated_at'], "%Y-%m-%dT%H:%M:%S.%fZ")

    def _validate_update_args(self, name: str, job: str) -> None:
        class UpdateArgs(BaseModel):
            name: str
            job: str

        UpdateArgs(name=name, job=job)

    def delete(self) -> 'User':
        url = self._build_url(f'/users/{self.id}')
        headers = self._get_headers()

        response = httpx.delete(url, headers=headers)
        self._handle_response(response)

        return self

    def login(self) -> str:
        try:
            self._validate_email(self.email)
        except ValidationError as e:
            raise ValueError(f"Invalid email: {e}")

        url = self._build_url('/token')
        headers = self._get_headers()
        json_data = {'email': self.email}

        response = httpx.post(url, headers=headers, json=json_data)
        return self._handle_response(response)['token']

    @classmethod
    def _validate_email(cls, email: str) -> None:
        class EmailModel(BaseModel):
            email: EmailStr

        EmailModel(email=email)

    def change(self, name: str, job: str) -> bytes:
        try:
            self._validate_update_args(name, job)
        except ValidationError as e:
            raise ValueError(f"Invalid arguments: {e}")

        url = self._build_url(f'/users/{self.id}')
        headers = self._get_headers()
        params = {'name': name, 'job': job}

        response = httpx.put(url, headers=headers, params=params)
        return response.content

This code is ~1.5 times bigger than the first and has no advantages over the code using the framework. He's just bigger. Some peoples would suggest ideas to reduce duplication, but in big projects, these ideas lead to bad architecture, especially when you need to implement both sync and async versions.

Also, you can handle API rate limiting, creating a RateLimit object.

router = Router('https://reqres.in/api', rate_limit=RateLimit(5, 1))

Along with OOP, you can use functional style:

from typing import Annotated
from sensei import Router, Path, APIModel

router = Router('https://pokeapi.co/api/v2/')


class Pokemon(APIModel):
    name: str
    id: int
    height: int
    weight: int
    types: list


@router.get('/pokemon/{pokemon_name}')
def get_pokemon(
        pokemon_name: Annotated[str, Path()],
) -> Pokemon:
    ...


pokemon = get_pokemon(pokemon_name="pikachu")
print(pokemon)

Installing sensei

To install sensei from PyPi, you can use that:

pip install sensei

To install sensei from GitHub, use that:

pip install git+https://github.com/CrocoFactory/sensei.git

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

sensei-0.1.0rc4.tar.gz (30.6 kB view details)

Uploaded Source

Built Distribution

sensei-0.1.0rc4-py3-none-any.whl (35.8 kB view details)

Uploaded Python 3

File details

Details for the file sensei-0.1.0rc4.tar.gz.

File metadata

  • Download URL: sensei-0.1.0rc4.tar.gz
  • Upload date:
  • Size: 30.6 kB
  • Tags: Source
  • Uploaded using Trusted Publishing? No
  • Uploaded via: twine/5.1.1 CPython/3.9.20

File hashes

Hashes for sensei-0.1.0rc4.tar.gz
Algorithm Hash digest
SHA256 5503ee7d26f509c1f0a3be00d6681b385aabc74512dd93a1939aeb9f8900c003
MD5 36f84bce72069de6a84af1ee96c004ef
BLAKE2b-256 73d177246ee19a9bdeb90eeda68b3061cee0ba61340bc48518d31ee7634ec0ab

See more details on using hashes here.

File details

Details for the file sensei-0.1.0rc4-py3-none-any.whl.

File metadata

  • Download URL: sensei-0.1.0rc4-py3-none-any.whl
  • Upload date:
  • Size: 35.8 kB
  • Tags: Python 3
  • Uploaded using Trusted Publishing? No
  • Uploaded via: twine/5.1.1 CPython/3.9.20

File hashes

Hashes for sensei-0.1.0rc4-py3-none-any.whl
Algorithm Hash digest
SHA256 35857d602ef8aad5f91a81eccbd0dd9040add9ea346348c2dcefbb4b5cd7bf4c
MD5 b76119a0a74dfa8e914b3c505a3a19b8
BLAKE2b-256 40a9a2bf8a33e464592b7870bab4541c94e52489262c6a8a7868f1ed2517fc5f

See more details on using hashes here.

Supported by

AWS AWS Cloud computing and Security Sponsor Datadog Datadog Monitoring Fastly Fastly CDN Google Google Download Analytics Microsoft Microsoft PSF Sponsor Pingdom Pingdom Monitoring Sentry Sentry Error logging StatusPage StatusPage Status page