"""
HTTP QUERY Method - RFC 10008
Ejemplos de uso en Python: requests, http.client y aiohttp
Autor: programacion.net
Licencia: MIT
"""
# ?????????????????????????????????????????????
# 1. Con requests (parcheo de método)
# ?????????????????????????????????????????????
# pip install requests
import requests
def query_con_requests(url: str, payload: dict, headers: dict = None) -> dict:
"""
Envía una petición HTTP QUERY usando requests.
requests no soporta QUERY de serie, pero acepta cualquier método
a través del parámetro method= en request().
"""
cabeceras = {
"Content-Type": "application/json",
"Accept": "application/json",
}
if headers:
cabeceras.update(headers)
respuesta = requests.request(
method="QUERY",
url=url,
json=payload,
headers=cabeceras,
)
respuesta.raise_for_status()
return respuesta.json()
# Ejemplo de uso
if __name__ == "__main__":
resultado = query_con_requests(
url="http://localhost:5000/api/productos/search",
payload={
"where": {
"categoria": "electronica",
"precio": {"gte": 100, "lte": 500},
"stock": {"gt": 0},
},
"sort": [{"field": "precio", "direction": "asc"}],
"limit": 20,
},
)
print("Resultados (requests):", resultado)
# ?????????????????????????????????????????????
# 2. Con http.client (biblioteca estándar)
# ?????????????????????????????????????????????
import http.client
import json as _json
def query_con_http_client(host: str, path: str, payload: dict, puerto: int = 80) -> dict:
"""
Envía una petición HTTP QUERY usando http.client de la biblioteca estándar.
http.client acepta cualquier cadena como método, incluido QUERY.
"""
cuerpo = _json.dumps(payload).encode("utf-8")
conexion = http.client.HTTPConnection(host, puerto)
conexion.request(
method="QUERY",
url=path,
body=cuerpo,
headers={
"Content-Type": "application/json",
"Accept": "application/json",
"Content-Length": str(len(cuerpo)),
},
)
respuesta = conexion.getresponse()
datos = _json.loads(respuesta.read().decode("utf-8"))
conexion.close()
return datos
# Ejemplo de uso
if __name__ == "__main__":
resultado = query_con_http_client(
host="localhost",
path="/api/productos/search",
payload={
"where": {"categoria": "electronica"},
"limit": 10,
},
)
print("Resultados (http.client):", resultado)
# ?????????????????????????????????????????????
# 3. Con aiohttp (async)
# ?????????????????????????????????????????????
# pip install aiohttp
import asyncio
import aiohttp
async def query_con_aiohttp(url: str, payload: dict) -> dict:
"""
Envía una petición HTTP QUERY de forma asíncrona con aiohttp.
aiohttp acepta cualquier verbo HTTP en el parámetro method=.
"""
async with aiohttp.ClientSession() as sesion:
async with sesion.request(
method="QUERY",
url=url,
json=payload,
headers={
"Content-Type": "application/json",
"Accept": "application/json",
},
) as respuesta:
respuesta.raise_for_status()
return await respuesta.json()
# Ejemplo de uso
async def main_aiohttp():
resultado = await query_con_aiohttp(
url="http://localhost:5000/api/productos/search",
payload={
"where": {"precio": {"lte": 200}},
"sort": [{"field": "nombre", "direction": "asc"}],
"limit": 5,
},
)
print("Resultados (aiohttp):", resultado)
if __name__ == "__main__":
asyncio.run(main_aiohttp())
# ?????????????????????????????????????????????
# 4. Servidor Flask que acepta QUERY
# ?????????????????????????????????????????????
# pip install flask
from flask import Flask, request, jsonify
app = Flask(__name__)
PRODUCTOS = [
{"id": 1, "nombre": "Teclado mecánico", "categoria": "electronica", "precio": 120, "stock": 15},
{"id": 2, "nombre": "Monitor 27"", "categoria": "electronica", "precio": 380, "stock": 8},
{"id": 3, "nombre": "Silla ergonómica", "categoria": "muebles", "precio": 450, "stock": 3},
{"id": 4, "nombre": "Ratón inalámbrico", "categoria": "electronica", "precio": 45, "stock": 30},
{"id": 5, "nombre": "Webcam HD", "categoria": "electronica", "precio": 95, "stock": 12},
]
def aplicar_filtros(items: list, where: dict) -> list:
"""Aplica los filtros del body QUERY sobre la lista de productos."""
resultado = items
for campo, condicion in where.items():
if isinstance(condicion, dict):
if "gte" in condicion:
resultado = [i for i in resultado if i.get(campo, 0) >= condicion["gte"]]
if "lte" in condicion:
resultado = [i for i in resultado if i.get(campo, 0) <= condicion["lte"]]
if "gt" in condicion:
resultado = [i for i in resultado if i.get(campo, 0) > condicion["gt"]]
if "lt" in condicion:
resultado = [i for i in resultado if i.get(campo, 0) < condicion["lt"]]
else:
resultado = [i for i in resultado if i.get(campo) == condicion]
return resultado
@app.route("/api/productos/search", methods=["QUERY", "POST"])
def buscar_productos():
"""
Acepta tanto QUERY (RFC 10008) como POST como fallback.
Devuelve productos filtrados según el body JSON.
"""
body = request.get_json(silent=True) or {}
where = body.get("where", {})
sort_rules = body.get("sort", [])
limit = body.get("limit", 50)
items = aplicar_filtros(PRODUCTOS, where)
for regla in reversed(sort_rules):
campo = regla.get("field", "id")
reverso = regla.get("direction", "asc") == "desc"
items = sorted(items, key=lambda x: x.get(campo, 0), reverse=reverso)
items = items[:limit]
# Cabecera Accept-Query: anunciamos qué formatos de consulta aceptamos
respuesta = jsonify({"results": items, "count": len(items), "method": request.method})
respuesta.headers["Accept-Query"] = "application/json"
return respuesta
@app.route("/api/productos/search", methods=["HEAD", "OPTIONS"])
def opciones_busqueda():
"""Descubrimiento: el cliente puede preguntar antes de hacer QUERY."""
resp = app.make_response("")
resp.headers["Accept-Query"] = "application/json"
resp.headers["Allow"] = "QUERY, POST, HEAD, OPTIONS"
return resp
if __name__ == "__main__":
# Para pruebas locales no usar en producción sin un servidor WSGI
app.run(debug=True, port=5000)
# ?????????????????????????????????????????????
# 5. Servidor FastAPI que acepta QUERY
# ?????????????????????????????????????????????
# pip install fastapi uvicorn
from fastapi import FastAPI, Request, Response
from fastapi.responses import JSONResponse
from pydantic import BaseModel
from typing import Optional
api = FastAPI(title="HTTP QUERY RFC 10008 - Demo")
class RangoNumerico(BaseModel):
gte: Optional[float] = None
lte: Optional[float] = None
gt: Optional[float] = None
lt: Optional[float] = None
class FiltrosBusqueda(BaseModel):
categoria: Optional[str] = None
precio: Optional[RangoNumerico] = None
stock: Optional[RangoNumerico] = None
class OrdenBusqueda(BaseModel):
field: str = "id"
direction: str = "asc"
class PeticionQuery(BaseModel):
where: Optional[FiltrosBusqueda] = None
sort: Optional[list[OrdenBusqueda]] = None
limit: Optional[int] = 50
@api.api_route("/api/productos/search", methods=["QUERY", "POST"])
async def buscar_productos_fastapi(peticion: Request) -> JSONResponse:
"""
Endpoint que acepta QUERY (RFC 10008) y POST como fallback.
FastAPI no registra QUERY en el router estándar, pero api_route()
con methods= acepta cualquier verbo HTTP.
"""
try:
body = await peticion.json()
params = PeticionQuery(**body)
except Exception:
params = PeticionQuery()
items = list(PRODUCTOS) # PRODUCTOS definido arriba
if params.where:
w = params.where
if w.categoria:
items = [i for i in items if i["categoria"] == w.categoria]
if w.precio:
if w.precio.gte is not None:
items = [i for i in items if i["precio"] >= w.precio.gte]
if w.precio.lte is not None:
items = [i for i in items if i["precio"] <= w.precio.lte]
if w.stock:
if w.stock.gt is not None:
items = [i for i in items if i["stock"] > w.stock.gt]
if params.sort:
for regla in reversed(params.sort):
items = sorted(items, key=lambda x: x.get(regla.field, 0),
reverse=(regla.direction == "desc"))
items = items[: params.limit]
respuesta = JSONResponse(
content={"results": items, "count": len(items), "method": peticion.method}
)
respuesta.headers["Accept-Query"] = "application/json"
return respuesta
# Arrancar con: uvicorn http_query_rfc10008:api --reload --port 8000
HTTP QUERY en Python: clientes requests, http.client, aiohttp y servidor Flask/FastAPI (RFC 10008)
Cinco ejemplos en Python para trabajar con el método HTTP QUERY del RFC 10008: clientes con requests, http.client y aiohttp, más un servidor Flask y otro FastAPI que lo aceptan junto a POST como fallback. Incluye filtros por rango, ordenación, cabecera Accept-Query y endpoint OPTIONS de descubrimiento.
Descargar adjuntos
COMPARTE ESTE TUTORIAL
COMPARTIR EN FACEBOOK
COMPARTIR EN TWITTER
COMPARTIR EN LINKEDIN
COMPARTIR EN WHATSAPP