Research
Security News
Malicious npm Packages Inject SSH Backdoors via Typosquatted Libraries
Socket’s threat research team has detected six malicious npm packages typosquatting popular libraries to insert SSH backdoors.
🙏 As a Python Backend developer, I've wasted so much time in recent years writing the same API clients over and over using Requests or HTTPX; At the same time, I could be so efficient by using FastAPI for API servers; I just wanted to save time for my upcoming projects, thinking that other developers might find it useful too.
Library to rapidly develop API clients in Python, based on Pydantic and Httpx, using almost only decorators and annotations.
✨ Main features:
Header
, Query
, Path
or Body
parameters.Body
with POST
-like opeations.httpx.Client
like you are used to, decorators simply build the httpx.Request
for you.async
operations, because httpx
and asyncio
are just amazingly fast.Install the library
pip install rapid-api-client
Declare your API endpoints using decorators and annotations, the method does not need any code, it will be generated by the decorator, just write ...
or pass
or whatever, it won't be called anyway 🙈.
class GithubIssuesApi(RapidApi):
@get("/repos/{owner}/{repo}/issues", response_class=TypeAdapter(List[Issue]))
def list_issues(self, owner: Annotated[str, Path()], repo: Annotated[str, Path()]): ...
@get("/repos/{owner}/{repo}/releases", response_class=TypeAdapter(List[Release]))
def list_releases(self, owner: Annotated[str, Path()], repo: Annotated[str, Path()]): ...
Use it
from httpx import Client
api = GithubIssuesApi(Client(base_url="https://api.github.com"))
issues = api.list_issues("essembeh", "rapid-api-client", state="closed")
for issue in issues:
print(f"Issue: {issue.title} [{issue.url}]")
Any HTTP method can be used with http
decorator
from rapid_api_client import RapidApi, http
class MyApi(RapidApi):
@http("GET", "/anything")
def get(self): ...
@http("POST", "/anything")
def post(self): ...
@http("DELETE", "/anything")
def delete(self): ...
Convenient decorators are available like get
, post
, delete
, put
, patch
from rapid_api_client import RapidApi, get, post, delete
class MyApi(RapidApi):
@get("/anything")
def get(self): ...
@post("/anything")
def post(self): ...
@delete("/anything")
def delete(self): ...
To use you API, you just need to instanciate it with a httpx.Client
like:
from httpx import Client
api = MyApi(Client(base_url="https://httpbin.org"))
resp = api.get()
resp.raise_for_status()
async
supportSince version
0.5.0
, the default client and annotations are synchronous,async
decorators and clients are inrapid_api_client.async_
package.
RapidApiClient support both sync
and async
methods, based on httpx.Client
and httpx.AsyncClient
.
To build an asynchronous client:
RapidApi
class with an AsyncClient
or use AsyncRapidApi
classget
or post
from rapid_api_client.async_
packageExample:
from rapid_api_client import RapidApi
from rapid_api_client.async_ import get, post, delete
from httpx import AsyncClient
class MyApi(RapidApi):
@get("/anything")
async def get(self): ...
@post("/anything")
async def post(self): ...
@delete("/anything")
async def delete(self): ...
# Usage
api = MyApi(AsyncClient(base_url="https://httpbin.org"))
resp = await api.get()
You can also use the dedicated AsyncRapidApi
class which provides a default factory for an AsyncClient
:
from rapid_api_client.async_ import get, post, delete, AsyncRapidApi
class MyApi(AsyncRapidApi):
@get("https://httpbin.org/anything")
async def get(self): ...
@post("https://httpbin.org/anything")
async def post(self): ...
@delete("https://httpbin.org/anything")
async def delete(self): ...
# Usage
api = MyApi()
resp = await api.get()
By default methods return a httpx.Response
object and the http return code is not tested (you have to call resp.raise_for_status()
if you need to ensure the response is OK).
But you can also specify a class so that the response is parsed, you can use:
httpx.Response
to get the response itself, this is the default behaviorstr
to get the response.text
bytes
to get the response.content
BaseModel
), the json will be automatically validatedBaseXmlModel
), the xml will be automatically validatedTypeAdapter
to parse the json, see pydantic docNote: When
response_class
is given (and is nothttpx.Response
), theraise_for_status()
is always called to ensure the http response is OK
class User(BaseModel): ...
class MyApi(RapidApi):
# this method return a httpx.Response
@get("/user/me")
def get_user_raw(self): ...
# this method returns a User class
@get("/user/me", response_class=User)
def get_user(self): ...
Like fastapi
you can use your method arguments to build the api path to call.
class MyApi(RapidApi):
# Path parameter
@get("/user/{user_id}")
def get_user(self, user_id: Annotated[int, Path()]): ...
# Path parameters with value validation
@get("/user/{user_id}")
def get_user(self, user_id: Annotated[PositiveInt, Path()]): ...
# Path parameters with a default value
@get("/user/{user_id}")
def get_user(self, user_id: Annotated[int, Path(default=1)]): ...
# Path parameters with a default value using a factory
@get("/user/{user_id}")
def get_user(self, user_id: Annotated[int, Path(default_factory=lambda: 42)]): ...
You can add query parameters
to your request using the Query
annotation.
class MyApi(RapidApi):
# Query parameter
@get("/issues")
def get_issues(self, sort: Annotated[str, Query()]): ...
# Query parameters with value validation
@get("/issues")
def get_issues(self, sort: Annotated[Literal["updated", "id"], Query()]): ...
# Query parameter with a default value
@get("/issues")
def get_issues(self, sort: Annotated[str, Query(default="updated")]): ...
# Query parameter with a default value using a factory
@get("/issues")
def get_issues(self, sort: Annotated[str, Query(default_factory=lambda: "updated")]): ...
# Query parameter with a default value
@get("/issues")
def get_issues(self, my_parameter: Annotated[str, Query(alias="sort")]): ...
You can add headers
to your request using the Header
annotation.
class MyApi(RapidApi):
# Header parameter
@get("/issues")
def get_issues(self, x_version: Annotated[str, Header()]): ...
# Header parameters with value validation
@get("/issues")
def get_issues(self, x_version: Annotated[Literal["2024.06", "2024.01"], Header()]): ...
# Header parameter with a default value
@get("/issues")
def get_issues(self, x_version: Annotated[str, Header(default="2024.06")]): ...
# Header parameter with a default value using a factory
@get("/issues")
def get_issues(self, x_version: Annotated[str, Header(default_factory=lambda: "2024.06")]): ...
# Header parameter with a default value
@get("/issues")
def get_issues(self, my_parameter: Annotated[str, Header(alias="x-version")]): ...
You can send a body with your request using the Body
annotation.
This body can be
Body
dict
object with JsonBody
PydanticBody
FileBody
class MyApi(RapidApi):
# send a string in request content
@post("/string")
def post_string(self, body: Annotated[str, Body()]): ...
# send a dict in request content as json
@post("/string")
def post_json(self, body: Annotated[dict, JsonBody()]): ...
# send a Pydantic model in serialized as json
@post("/model")
def post_model(self, body: Annotated[MyPydanticClass, PydanticBody()]): ...
# send a multiple files
@post("/files")
def post_files(self, report: Annotated[bytes, FileBody()], image: Annotated[bytes, FileBody()]): ...
# send a form
@post("/form")
def post_form(self, my_param: Annotated[str, FormBody(alias="name")], extra_fields: Annotated[Dict[str, str], FormBody()]): ...
Xml is also supported is you use Pydantic-Xml, either for responses with response_class
or for POST/PUT content with PydanticXmlBody
.
class ResponseXmlRootModel(BaseXmlModel): ...
class MyApi(RapidApi):
# parse response xml content
@get("/get", response_class=ResponseXmlRootModel)
def get_xml(self): ...
# serialize xml model automatically
@post("/post")
def post_xml(self, body: Annotated[ResponseXmlRootModel, PydanticXmlBody()]): ...
See example directory for some examples
FAQs
Rapidly develop your API clients using decorators and annotations
We found that rapid-api-client demonstrated a healthy version release cadence and project activity because the last version was released less than a year ago. It has 1 open source maintainer collaborating on the project.
Did you know?
Socket for GitHub automatically highlights issues in each pull request and monitors the health of all your open source dependencies. Discover the contents of your packages and block harmful activity before you install or update your dependencies.
Research
Security News
Socket’s threat research team has detected six malicious npm packages typosquatting popular libraries to insert SSH backdoors.
Security News
MITRE's 2024 CWE Top 25 highlights critical software vulnerabilities like XSS, SQL Injection, and CSRF, reflecting shifts due to a refined ranking methodology.
Security News
In this segment of the Risky Business podcast, Feross Aboukhadijeh and Patrick Gray discuss the challenges of tracking malware discovered in open source softare.