创建简单的Python微服务Demo与FastAPI
在微服务架构中,通过FastAPI框架创建一个简单的Python微服务Demo涉及多个步骤,包括定义服务、使用框架、进行通信等。在这篇文章中,我们将使用FastAPI框架创建两个简单的微服务,它们通过RESTful API进行通信。
首先,确保你已经安装了FastAPI和uvicorn,可以使用以下命令安装:
pip install fastapi uvicorn
然后,我们创建两个微服务:
微服务1:用户服务 (user_service.py
)
from fastapi import FastAPIapp = FastAPI()@app.get("/users/{user_id}")
async def read_user(user_id: int):return {"user_id": user_id, "username": f"user{user_id}"}
微服务2:订单服务 (order_service.py
)
from fastapi import FastAPI, HTTPExceptionapp = FastAPI()@app.get("/orders/{order_id}")
async def read_order(order_id: int):if order_id <= 0:raise HTTPException(status_code=400, detail="Invalid order ID")return {"order_id": order_id, "status": "processed"}
然后,我们可以使用uvicorn
命令来运行这两个微服务:
uvicorn user_service:app --host 0.0.0.0 --port 8000 --reload
uvicorn order_service:app --host 0.0.0.0 --port 8001 --reload
现在,你有两个微服务在不同的端口上运行。user_service
提供了获取用户信息的接口,而order_service
提供了获取订单信息的接口。
微服务通信:
我们可以使用httpx
库来模拟微服务之间的通信。确保你已经安装了它:
pip install httpx
然后,可以创建一个简单的脚本 (communication_demo.py
) 来演示微服务之间的通信:
import httpxasync def get_user(user_id):async with httpx.AsyncClient() as client:response = await client.get(f"http://127.0.0.1:8000/users/{user_id}")return response.json()async def get_order(order_id):async with httpx.AsyncClient() as client:response = await client.get(f"http://127.0.0.1:8001/orders/{order_id}")return response.json()async def main():user_info = await get_user(1)print("User Info:", user_info)order_info = await get_order(1001)print("Order Info:", order_info)if __name__ == "__main__":import asyncioasyncio.run(main())
在这个演示中,communication_demo.py
脚本通过调用 get_user
和 get_order
函数来获取用户和订单信息。你可以运行这个脚本,看到微服务之间的通信效果。
这只是一个简单的微服务架构的Demo,实际的微服务架构可能涉及到更多的复杂性,如服务注册与发现、负载均衡、安全性等。