2019-9-14 331 0
2019-9-14 288 0
2020-6-4 8 0
Python极客

安装&加载pip3installpymongo-ihttps://mirrors.aliyun.com/pypi/simple/importpymongo连接MongoDB服务client=pymongo.MongoClient("mongodb://localhost:27017/")print(client)--------------#Mongo服务MongoClient(host=['localhost:27017'],document_class=dict,tz_aware=False,connect=True)获取所有数据库dblist=client.list_database_names()print(dblist)-------------#所有数据库名称['admin','config','local']创建数据库&获取数据库对象如果没有该数据库则创建db=client["db"]print(db)---------#db数据库对象Database(MongoClient(host=['localhost:27017'],document_class=dict,tz_aware=False,connect=True),'db')创建集合&获取集合如果没有该集合则创建#创建集合collection=db["collection"]print(sets)-----------#sets集合对象Collection(Database(MongoClient(host=['localhost:27017'],document_class=dict,tz_aware=False,connect=True),'db'),'sets')#获取所有集合sets=db.list_collection_names()print(sets)--------------#该数据库所有集合['collection']插入数据#插入一条数据data={"name":"bigdataboy","age":"18"}x=collection.insert_one(data)print(x.inserted_id)--------#数据的_id5ed666e1bca2037c30662e97#插入多条数据datas=[{"name":"大数据男孩","age":"18"},{"name":"bigdataboy","age":"16","addr":"China"},{"name":"bigdataboy","age":"18"}]x=collection.insert_many(datas)print(x.inserted_ids)---------------------#插入数据的_id[ObjectId('5ed667e044fb69d445e510d8'),ObjectId('5ed667e044fb69d445e510d9'),ObjectId('5ed667e044fb69d445e510da')]自定义_id#链式写法collection=pymongo.MongoClient("mongodb://localhost:27017/")["db"]["collection"]#数据自行固定_iddatas=[{"_id":"1","name":"大数据男孩","age":"18"},{"_id":"2","name":"bigdataboy","age":"16","addr":"China"},{"_id":"3","name":"bigdataboy","age":"18"}]i=collection.insert_many(datas)print(i.inserted_ids)---------------------#插入数据的_id['1','2','3']查看所有数据collection=pymongo.MongoClient("mongodb://localhost:27017/")["db"]["collection"]forxincollection.find():print(x)------------#所有数据{'_id':'1','name':'大数据男孩','age':'18'}{'_id':'2','name':'bigdataboy','age':'16','addr':'China'}{'_id':'3','name':'bigdataboy','age':'18'}

2020-6-3 6 0
Python极客

说明在许多情况下,您需要将错误通知给正在使用API​​的客户端。FastAPI使用Python的HTTPException异常进行处理,所以raise它。实例fromfastapiimportFastAPI,HTTPExceptionimportuvicornapp=FastAPI()@app.get("/user")asyncdefread_item(age:int,name:str):ifage<18:#返回客户端500状态码及错误原因raiseHTTPException(status_code=500,detail="age小于18")return{"name":name,"age":age}if__name__=='__main__':uvicorn.run(app=app,host="127.0.0.1",port=80,)添加自定义Headers@app.get("/user")asyncdefread_item(age:int,name:str):ifage<18:#返回客户端500状态码及错误原因raiseHTTPException(status_code=500,#detail可以是list、dict、strdetail={"msg":"age小于18"},#自定义headersheaders={"X-Error":"Theregoesmyerror"})return{"name":name,"age":age}自定义异常处理程序fromfastapiimportFastAPI,Requestfromfastapi.responsesimportJSONResponseimportuvicornapp=FastAPI()#异常参数类classUnicornException(Exception):def__init__(self,status_code:int,msg:str):#错误状态码self.status_code=status_code#错误内容self.msg=msg#异常响应@app.exception_handler(UnicornException)asyncdefunicorn_exception_handler(request:Request,exc:UnicornException):returnJSONResponse(#错误响应状态码status_code=exc.status_code,#错误返回内容格式content={"msg":exc.msg},)@app.get("/user/{name}")asyncdefread_item(name:str):ifname=="bigdata":#错误提示raiseUnicornException(status_code=404,msg="用户名错误,没有该用户")return{"name":name}if__name__=='__main__':uvicorn.run(app=app,host="127.0.0.1",port=80,)测试URL:http://127.0.0.1/user/bigdata?name=bigdataboy

Python极客

实例代码fromfastapiimportFastAPIimportuvicornapp=FastAPI()@app.post("/user/{name}")#路径参数asyncdefread_item(name:str):return{"name":name}if__name__=='__main__':uvicorn.run(app=app,host="127.0.0.1",port=80,)导入验证fromfastapiimportFastAPI,Query可选参数name可传可不传#正常写法name为可选参数@app.get("/user/")asyncdefread_item(age:int,name:str=None):return{"name":name,"age":age}#验证写法name为可选参数@app.get("/user/")asyncdefread_item(age:int,name:str=Query(None)):return{"name":name,"age":age}必选参数name必须传#正常写法name为必选参数asyncdefread_item(age:int,name:str):return{"name":name,"age":age}#验证写法name为必选参数asyncdefread_item(age:int,name:str=Query(...)):return{"name":name,"age":age}默认参数值#正常写法name默认值asyncdefread_item(age:int,name:str="bigdataboy"):return{"name":name,"age":age}#验证写法name默认值asyncdefread_item(age:int,name:str=Query("bigdataboy")):return{"name":name,"age":age}Query验证Query主要支持一下验证长度验证:max_length、min_length正则验证:regex@app.get("/user")#路径参数asyncdefread_item(age:int,name:str=Query(...,#该参数必选min_length=1,#最小长度为1max_length=50,#最大长度为50#regex=re.compile(".+bigdataboy.+").pattern#使用编译的正则表达式需要导入re库regex=".+bigdataboy.+"#正则表达式匹配验证)):return{"name":name,"age":age}其他参数这些参数主要是对接口进行描述@app.get("/user")#路径参数asyncdefread_item(age:int,name:str=Query(...,#该参数必选title="name",#标题description="用户的新名字",#参数作用的描述alias="new-name",#接口参数别名,URL参数就使用该别名deprecated=True,#代表该参数,即将弃用)):return{"name":name,"age":age}

2020-5-27 30 0
Python极客

路径参数比如访问:127.0.0.1/user/123/fromfastapiimportFastAPIimportuvicornapp=FastAPI()@app.get("/user/{id}/")asyncdefmain(id:int):return{"id":id}if__name__=='__main__':uvicorn.run(app=app,host="127.0.0.1",port=80,)API文档地址:http://127.0.0.1/docs固定的路径参数@app.get("/user/id/")asyncdefmain():return{"id":"thecurrentuser"}@app.get("/user/{id}/")asyncdefmain(id:int):return{"id":id}预留路径值传入的路径参数只能在预留路径里,不然就报错fromfastapiimportFastAPI#导入枚举类fromenumimportEnumimportuvicorn#预留路径类继承str,文档将能够知道这些值的类型classModelName(str,Enum):#预留的路径alexnet="alexnet"resnet="resnet"lenet="lenet"app=FastAPI()@app.get("/model/{model_name}")asyncdefget_model(model_name:ModelName):#参数类型是预留路径类#第一种判定写法ifmodel_name==ModelName.alexnet:return{"model_name":model_name,"message":"DeepLearningFTW!"}#第二种判定写法ifmodel_name.value=="lenet":return{"model_name":model_name,"message":"LeCNNalltheimages"}#默认返回return{"model_name":model_name,"message":"Havesomeresiduals"}if__name__=='__main__':uvicorn.run(app=app,host="127.0.0.1",port=80,)路径转换器访问/files/bigdataboy/a.txt,file_path将能得到bigdataboy/a.txt#添加上:path即可@app.get("/files/{file_path:path}")asyncdefread_file(file_path:str):return{"file_path":file_path}查询参数查询参数是指?后面的&分割的键值对http://127.0.0.1:8000/items/?skip=0&limit=2fromfastapiimportFastAPIimportuvicornapp=FastAPI()fake_items_db=[{"item_name":"Foo"},{"item_name":"Bar"},{"item_name":"Baz"}]@app.get("/items/")#设置的默认值,当默认值是None时,表示该值为可选asyncdefread_item(skip:int=0,limit:int=2,default:str=None):#一个列表切片returnfake_items_db[skip:skip+limit]if__name__=='__main__':uvicorn.run(app=app,host="127.0.0.1",port=80,)注意如果声明可选时出现以下类似错误#声明可选limit:int=None#出现以下类似错误Incompatibletypesinassignment(expressionhastype"None",variablehastype"int")这是你需要这样做fromtypingimportOptionallimit:Optional[int]=None完整实例fromfastapiimportFastAPIfromtypingimportOptionalimportuvicornapp=FastAPI()@app.get("/items/")asyncdefread_item(skip:str,limit:Optional[int]=None):item={"skip":skip,"limit":limit}returnitemif__name__=='__main__':uvicorn.run(app=app,host="127.0.0.1",port=80,)

2020-5-26 22 0
2020-5-21 19 0