Security News
Research
Data Theft Repackaged: A Case Study in Malicious Wrapper Packages on npm
The Socket Research Team breaks down a malicious wrapper package that uses obfuscation to harvest credentials and exfiltrate sensitive data.
fastapi-microservice-helper
Advanced tools
FastAPI Microservice Helper is an SDK designed to facilitate communication between microservices over HTTP.
Let's say we have a CoreMicroservice that handles creating and retrieving posts. Now, we want to call this CoreMicroservice from a UserMicroservice.
In the CoreMicroservice, you define actions for handling posts. Here's an example:
# http/microservices/post.py
from uuid import UUID
from fastapi import Depends, Query
from fastapi_microservice_helper import action, microservice
from fastapi_request_helper.decorators.http_method import response, tag
from posts.http.responses.general_post_response import GeneralPostResponse
from posts.services.post import PostService
class GeneralPostResponse(BaseModel):
id: UUID
author_id: UUID
class CreatePostDto(BaseDto):
author_id: UUID
@microservice()
@tag('Microservices Post')
class PostMicroservice:
def __init__(self, post_service: PostService = Depends()):
self.post_service = post_service
@action()
@response(GeneralPostResponse)
async def get_post(self, post_id: UUID):
return await self.post_service.find_general_post_by_id(post_id)
@action()
async def create_post(self, dto: CreatePostDto):
await self.post_service.increment_post_likes(dto)
In this example, we define two actions for PostMicroservice:
get_post to retrieve a post by ID create_post to create a new post.
FastAPI Microservice Helper automatically generates the SDK code based on the server definitions. Here’s an example of what the generated client SDK might look like:
# sdk/src/core_microservice/sdk.py
...
class BaseMicroserviceClient:
def filter_none_values(self, query_params: dict | None):
return {key: value for key, value in query_params.items() if value is not None} if query_params else None
async def send(
self, url: str, query_params: dict, body_params: any, response_type: any, option: MicroserviceOption = None
):
if not CoreMicroserviceConfig.url:
raise Exception('Please config microservice url')
url = CoreMicroserviceConfig.url + url
if not option:
option = MicroserviceOption()
async with httpx.AsyncClient() as client:
response = await client.post(
url=url,
headers=option.headers,
params=self.filter_none_values(query_params),
data=body_params if not option.is_json else None,
json=Normailization.normalize(body_params) if option.is_json else None,
)
data = response.json()
if response.status_code < 200 or response.status_code > 299:
raise HTTPException(status_code=response.status_code, detail=data)
if not response_type:
return data
return TypeAdapter(response_type).validate_python(data)
class PostMicroservice(BaseMicroserviceClient):
async def decrement_post_likes(self, dto: DecrementPostLikesDto , option: MicroserviceOption = None) -> None:
return await self.send('/microservices/PostMicroservice/decrement_post_likes', None, dto, None, option )
async def get_post(self, post_id: UUID , option: MicroserviceOption = None) -> GeneralPostResponse:
return await self.send('/microservices/PostMicroservice/get_post', {'post_id': post_id}, None, GeneralPostResponse, option)
async def create_post(self, dto: CreatePostDto , option: MicroserviceOption = None) -> None:
return await self.send('/microservices/PostMicroservice/create_post', None, dto, None, option)
Increase the version in sdk.toml
file then build and publish it, for example, publish the sdk as core-microservice==0.0.1
.
Add the SDK to your project’s dependencies in requirements.txt:
# requirements.txt
core-microservice==0.0.1
Then install the required packages:
pip install -r requirements.txt
Set the base URL for the CoreMicroservice:
# config/microservice.py
from core_microservice import CoreMicroserviceConfig
CoreMicroserviceConfig.url = 'http://127.0.0.1:8000'
In the UserMicroservice, you can now use the SDK to call actions from CoreMicroservice:
# services/user.py
class UserService:
def __init__(self, post_microservice: PostMicroservice = Depends()):
self.post_microservice = post_microservice
async def create_post(self, dto: CreatePostDto):
await self.post_microservice.create_post(dto)
async def get_post(self, post_id: UUID):
return await self.post_microservice.get_post(post_id)
FAQs
FastAPI Microservice Helper
We found that fastapi-microservice-helper 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.
Security News
Research
The Socket Research Team breaks down a malicious wrapper package that uses obfuscation to harvest credentials and exfiltrate sensitive data.
Research
Security News
Attackers used a malicious npm package typosquatting a popular ESLint plugin to steal sensitive data, execute commands, and exploit developer systems.
Security News
The Ultralytics' PyPI Package was compromised four times in one weekend through GitHub Actions cache poisoning and failure to rotate previously compromised API tokens.