怎么if __name__ == "__main__":办？
# Threading example
import time, thread
def myfunction(string, sleeptime, lock, *args):
while True:
lock.acquire()
time.sleep(sleeptime)
lock.release()
time.sleep(sleeptime)
if __name__ == "__main__":
lock = thread.allocate_lock()
thread.start_new_thread(myfunction, ("Thread #: 1", 2, lock))
thread.start_new_thread(myfunction, ("Thread #: 2", 2, lock))
"coroutine"和"thread"之间有什么区别？
我一直在修改 Flask 和 FastAPI 以了解它如何充当服务器。
我想知道的主要事情之一是 Flask 和 FastAPI 如何处理来自多个客户端的多个请求。
特别是当代码有效率问题时（数据库查询时间长）。
所以，我尝试制作一个简单的代码来理解这个问题。
代码很简单，当客户端访问路由时，应用程序休眠10秒才返回结果。
它看起来像这样：
快速API
import uvicorn
from fastapi import FastAPI
from time import sleep
app = FastAPI()
@app.get('/')
async def root():
print('Sleeping for 10')
sleep(10)
print('Awake')
return {'message': 'hello'}
if __name__ == "__main__":
uvicorn.run(app, host="127.0.0.1", port=8000)
烧瓶
from flask import Flask
from flask_restful import Resource, Api
from time import sleep
app = Flask(__name__)
api = Api(app)
class Root(Resource):
def get(self):
print('Sleeping for 10')
sleep(10)
print('Awake')
return {'message': 'hello'}
api.add_resource(Root, …在同步而非非模式下使用 FastAPI async，我希望能够接收 POST 请求的原始、未更改的正文。
我能找到的所有示例都显示async代码，当我以正常同步方式尝试时，它们request.body()显示为协程对象。
XML当我通过向此端点发布一些内容来测试它时，我得到了一个500 "Internal Server Error".
from fastapi import FastAPI, Response, Request, Body
app = FastAPI()
@app.get("/")
def read_root():
return {"Hello": "World"}
@app.post("/input")
def input_request(request: Request):
# how can I access the RAW request body here?
body = request.body()
# do stuff with the body here
return Response(content=body, media_type="application/xml")
这对于 FastAPI 来说是不可能的吗？
注意：简化的输入请求如下所示：
POST http://127.0.0.1:1083/input
Content-Type: application/xml
<XML>
<BODY>TEST</BODY>
</XML>
我无法控制输入请求的发送方式，因为我需要替换现有的 SOAP API。
在处理 FastAPI 请求时，我有一个受 CPU 限制的任务要对列表的每个元素执行。我想在多个 CPU 内核上进行此处理。
在 FastAPI 中执行此操作的正确方法是什么？我可以使用标准multiprocessing模块吗？到目前为止，我发现的所有教程/问题仅涵盖 I/O 绑定任务，如 Web 请求。
我正在使用 FastAPI 在 python 中创建一个服务器，并且我想要一个与我的 API 无关的函数，每 5 分钟在后台运行一次（例如从 API 检查内容并根据响应打印内容）
我尝试创建一个运行该函数的线程start_worker，但它不打印任何内容。
有谁知道该怎么做？
def start_worker():
print('[main]: starting worker...')
my_worker = worker.Worker()
my_worker.working_loop() # this function prints "hello" every 5 seconds
if __name__ == '__main__':
print('[main]: starting...')
uvicorn.run(app, host="0.0.0.0", port=8000, reload=True)
_worker_thread = Thread(target=start_worker, daemon=False)
_worker_thread.start()
在创建我需要的实际应用程序之前,我试图对此有一个基本的了解.我最近从2.7移到了3.3.
从python文档直接复制粘贴此代码失败,这里稍微简单的示例也是如此.
这是我的代码,派生自第二个例子:
import concurrent.futures
nums = [1,2,3,4,5,6,7,8,9,10]
def f(x):
return x * x
# Make sure the map and function are working
print([val for val in map(f, nums)])
# Test to make sure concurrent map is working
with concurrent.futures.ProcessPoolExecutor() as executor:
for item in executor.map(f, nums):
print(item)
这是输出:
[1, 4, 9, 16, 25, 36, 49, 64, 81, 100]
Traceback (most recent call last):
File "<string>", line 420, in run_nodebug
File "<module1>", line 13, in …我正在构建一个简单的 API 来测试数据库。当我使用 get request 时一切正常，但如果我更改为 post，我会收到“无法处理的实体”错误：
这是 FastAPI 代码：
from fastapi import FastAPI
app = FastAPI()
@app.post("/")
def main(user):
return user
然后，我的请求使用 javascript
let axios = require('axios')
data = {
user: 'smith'
}
axios.post('http://localhost:8000', data)
.then(response => (console.log(response.url)))
并使用 Python
import requests
url = 'http://127.0.0.1:8000'
data = {'user': 'Smith'}
response = requests.post(url, json=data)
print(response.text)
我也尝试解析为 json，使用 utf-8 编码，并更改标题。没有什么对我有用。
我有一个相对简单的 FastAPI 应用程序，它接受查询并从 ChatGPT 的 API 流回响应。ChatGPT 正在流回结果，我可以看到它在输入时被打印到控制台。
不工作的是StreamingResponse通过 FastAPI 返回。相反，响应会一起发送。我真的不知道为什么这不起作用。
这是 FastAPI 应用程序代码：
import os
import time
import openai
import fastapi
from fastapi import Depends, HTTPException, status, Request
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from fastapi.responses import StreamingResponse
auth_scheme = HTTPBearer()
app = fastapi.FastAPI()
openai.api_key = os.environ["OPENAI_API_KEY"]
def ask_statesman(query: str):
#prompt = router(query)
completion_reason = None
response = ""
while not completion_reason or completion_reason == "length":
openai_stream = openai.ChatCompletion.create(
model="gpt-3.5-turbo",
messages=[{"role": "user", "content": query}],
temperature=0.0,
stream=True,
)
for …python ×8
fastapi ×6
coroutine ×2
uvicorn ×2
architecture ×1
asgi ×1
axios ×1
flask ×1
idioms ×1
namespaces ×1
openai-api ×1
python-3.x ×1
starlette ×1
streaming ×1
system ×1
terminology ×1
wsgi ×1