Ir para o conteúdo

Guia de Início Rápido - Backend Python

Instalação

uv add bfast-py
# ou
pip install bfast-py

Uso Básico

Serialização Simples

import b_fast

# Seus dados
data = [{"id": i, "name": f"User {i}"} for i in range(1000)]

# 1-liner direto (compressão ativada por padrão)
encoded = b_fast.encode(data)
print(f"Tamanho: {len(encoded)} bytes")

# 1-liner de deserialização (descompressão automática por padrão)
decoded = b_fast.decode(encoded)

# Ou através da instância da classe BFast
encoder = b_fast.BFast()
encoded = encoder.encode_packed(data)      # compress=True por padrão
decoded = encoder.decode_packed(encoded)  # decompress=True por padrão

Compressão

Por padrão, o B-FAST aplica compressão automaticamente (compress=True), economizando até 89% de tráfego de rede.

Para desativar a compressão em micro-payloads (< 1KB) ou para latência ultra-baixa de CPU: 

# Serialização sem compressão
raw_binary = b_fast.encode(data, compress=False)
# ou
raw_binary = encoder.encode_packed(data, compress=False)

Integração com FastAPI ⭐ Recomendado

O B-FAST disponibiliza classes nativas de Response e StreamingResponse para FastAPI e Starlette.

Resposta Padrão

from fastapi import FastAPI
from pydantic import BaseModel
from b_fast import BFastResponse

app = FastAPI()

class User(BaseModel):
    id: int
    name: str
    email: str

@app.get("/users", response_class=BFastResponse)
async def get_users():
    return [User(id=i, name=f"User {i}", email=f"user{i}@example.com") for i in range(1000)]

Resposta Streamable HTTP (Streaming Contínuo) 🌊

Transmita frames binários em tempo real sobre HTTP/1.1 (Chunked), HTTP/2 ou HTTP/3:

import asyncio
from fastapi import FastAPI
from b_fast import BFastStreamingResponse

app = FastAPI()

@app.get("/stream-users", response_class=BFastStreamingResponse)
async def stream_users():
    async def user_generator():
        for i in range(100):
            yield {"id": i, "name": f"User {i}", "status": "active"}
            await asyncio.sleep(0.05)

    return user_generator()

Integração com FastMCP 2.0 🤖

Transmita grandes massas de dados de ferramentas de IA utilizando decoradores B-FAST, reduzindo o consumo de tokens em até 85%:

from b_fast.fastmcp import FastMCPBFast, bfast_tool

mcp = FastMCPBFast("Servidor de Analytics")

@mcp.tool()
@bfast_tool(compress=True)
def consulta_dataset(limite: int = 1000) -> list[dict]:
    return [{"id": i, "valor": i * 1.5} for i in range(limite)]

Próximos Passos