35 changed files with 300 additions and 311 deletions
@ -0,0 +1,17 @@ |
|||||
|
from fastapi import Body, FastAPI |
||||
|
from pydantic import BaseModel, Schema |
||||
|
|
||||
|
app = FastAPI() |
||||
|
|
||||
|
|
||||
|
class Item(BaseModel): |
||||
|
name: str |
||||
|
description: str = Schema(None, title="The description of the item", max_length=300) |
||||
|
price: float = Schema(..., gt=0, description="The price must be greater than zero") |
||||
|
tax: float = None |
||||
|
|
||||
|
|
||||
|
@app.put("/items/{item_id}") |
||||
|
async def update_item(*, item_id: int, item: Item = Body(..., embed=True)): |
||||
|
results = {"item_id": item_id, "item": item} |
||||
|
return results |
@ -1,8 +1,17 @@ |
|||||
from fastapi import Cookie, FastAPI |
from typing import List |
||||
|
|
||||
|
from fastapi import FastAPI |
||||
|
from pydantic import BaseModel |
||||
|
from pydantic.types import UrlStr |
||||
|
|
||||
app = FastAPI() |
app = FastAPI() |
||||
|
|
||||
|
|
||||
@app.get("/items/") |
class Image(BaseModel): |
||||
async def read_items(*, ads_id: str = Cookie(None)): |
url: UrlStr |
||||
return {"ads_id": ads_id} |
name: str |
||||
|
|
||||
|
|
||||
|
@app.post("/images/multiple/") |
||||
|
async def create_multiple_images(*, images: List[Image]): |
||||
|
return images |
||||
|
@ -1,8 +1,8 @@ |
|||||
from fastapi import FastAPI, Header |
from fastapi import Cookie, FastAPI |
||||
|
|
||||
app = FastAPI() |
app = FastAPI() |
||||
|
|
||||
|
|
||||
@app.get("/items/") |
@app.get("/items/") |
||||
async def read_items(*, accept_encoding: str = Header(None)): |
async def read_items(*, ads_id: str = Cookie(None)): |
||||
return {"Accept-Encoding": accept_encoding} |
return {"ads_id": ads_id} |
||||
|
@ -1,19 +1,8 @@ |
|||||
from typing import Set |
from fastapi import FastAPI, Header |
||||
|
|
||||
from fastapi import FastAPI |
|
||||
from pydantic import BaseModel |
|
||||
|
|
||||
app = FastAPI() |
app = FastAPI() |
||||
|
|
||||
|
|
||||
class Item(BaseModel): |
@app.get("/items/") |
||||
name: str |
async def read_items(*, strange_header: str = Header(None, convert_underscores=False)): |
||||
description: str = None |
return {"strange_header": strange_header} |
||||
price: float |
|
||||
tax: float = None |
|
||||
tags: Set[str] = [] |
|
||||
|
|
||||
|
|
||||
@app.post("/items/", response_model=Item) |
|
||||
async def create_item(*, item: Item): |
|
||||
return item |
|
||||
|
@ -1,18 +1,19 @@ |
|||||
|
from typing import Set |
||||
|
|
||||
from fastapi import FastAPI |
from fastapi import FastAPI |
||||
from pydantic import BaseModel |
from pydantic import BaseModel |
||||
from pydantic.types import EmailStr |
|
||||
|
|
||||
app = FastAPI() |
app = FastAPI() |
||||
|
|
||||
|
|
||||
class UserIn(BaseModel): |
class Item(BaseModel): |
||||
username: str |
name: str |
||||
password: str |
description: str = None |
||||
email: EmailStr |
price: float |
||||
full_name: str = None |
tax: float = None |
||||
|
tags: Set[str] = [] |
||||
|
|
||||
|
|
||||
# Don't do this in production! |
@app.post("/items/", response_model=Item) |
||||
@app.post("/user/", response_model=UserIn) |
async def create_item(*, item: Item): |
||||
async def create_user(*, user: UserIn): |
return item |
||||
return user |
|
||||
|
@ -1,8 +1,40 @@ |
|||||
from fastapi import FastAPI, Form |
from fastapi import FastAPI |
||||
|
from pydantic import BaseModel |
||||
|
from pydantic.types import EmailStr |
||||
|
|
||||
app = FastAPI() |
app = FastAPI() |
||||
|
|
||||
|
|
||||
@app.post("/login/") |
class UserBase(BaseModel): |
||||
async def login(*, username: str = Form(...), password: str = Form(...)): |
username: str |
||||
return {"username": username} |
email: EmailStr |
||||
|
full_name: str = None |
||||
|
|
||||
|
|
||||
|
class UserIn(UserBase): |
||||
|
password: str |
||||
|
|
||||
|
|
||||
|
class UserOut(UserBase): |
||||
|
pass |
||||
|
|
||||
|
|
||||
|
class UserInDB(UserBase): |
||||
|
hashed_password: str |
||||
|
|
||||
|
|
||||
|
def fake_password_hasher(raw_password: str): |
||||
|
return "supersecret" + raw_password |
||||
|
|
||||
|
|
||||
|
def fake_save_user(user_in: UserIn): |
||||
|
hashed_password = fake_password_hasher(user_in.password) |
||||
|
user_in_db = UserInDB(**user_in.dict(), hashed_password=hashed_password) |
||||
|
print("User saved! ..not really") |
||||
|
return user_in_db |
||||
|
|
||||
|
|
||||
|
@app.post("/user/", response_model=UserOut) |
||||
|
async def create_user(*, user_in: UserIn): |
||||
|
user_saved = fake_save_user(user_in) |
||||
|
return user_saved |
||||
|
@ -1,8 +1,8 @@ |
|||||
from fastapi import FastAPI, File |
from fastapi import FastAPI, Form |
||||
|
|
||||
app = FastAPI() |
app = FastAPI() |
||||
|
|
||||
|
|
||||
@app.post("/files/") |
@app.post("/login/") |
||||
async def create_file(*, file: bytes = File(...)): |
async def login(*, username: str = Form(...), password: str = Form(...)): |
||||
return {"file_size": len(file)} |
return {"username": username} |
||||
|
@ -1,8 +1,8 @@ |
|||||
from fastapi import FastAPI, File, Form |
from fastapi import FastAPI, File |
||||
|
|
||||
app = FastAPI() |
app = FastAPI() |
||||
|
|
||||
|
|
||||
@app.post("/files/") |
@app.post("/files/") |
||||
async def create_file(*, file: bytes = File(...), token: str = Form(...)): |
async def create_file(*, file: bytes = File(...)): |
||||
return {"file_size": len(file), "token": token} |
return {"file_size": len(file)} |
||||
|
@ -1,20 +1,8 @@ |
|||||
from typing import Set |
from fastapi import FastAPI, File, Form |
||||
|
|
||||
from fastapi import FastAPI |
|
||||
from pydantic import BaseModel |
|
||||
from starlette.status import HTTP_201_CREATED |
|
||||
|
|
||||
app = FastAPI() |
app = FastAPI() |
||||
|
|
||||
|
|
||||
class Item(BaseModel): |
@app.post("/files/") |
||||
name: str |
async def create_file(*, file: bytes = File(...), token: str = Form(...)): |
||||
description: str = None |
return {"file_size": len(file), "token": token} |
||||
price: float |
|
||||
tax: float = None |
|
||||
tags: Set[str] = [] |
|
||||
|
|
||||
|
|
||||
@app.post("/items/", response_model=Item, status_code=HTTP_201_CREATED) |
|
||||
async def create_item(*, item: Item): |
|
||||
return item |
|
||||
|
@ -1,8 +1,33 @@ |
|||||
|
from typing import Set |
||||
|
|
||||
from fastapi import FastAPI |
from fastapi import FastAPI |
||||
|
from pydantic import BaseModel |
||||
|
|
||||
app = FastAPI() |
app = FastAPI() |
||||
|
|
||||
|
|
||||
@app.get("/items/", deprecated=True) |
class Item(BaseModel): |
||||
async def read_items(): |
name: str |
||||
return [{"item_id": "Foo"}] |
description: str = None |
||||
|
price: float |
||||
|
tax: float = None |
||||
|
tags: Set[str] = [] |
||||
|
|
||||
|
|
||||
|
@app.post( |
||||
|
"/items/", |
||||
|
response_model=Item, |
||||
|
summary="Create an item", |
||||
|
response_description="The created item", |
||||
|
) |
||||
|
async def create_item(*, item: Item): |
||||
|
""" |
||||
|
Create an item with all the information: |
||||
|
|
||||
|
* name: each item must have a name |
||||
|
* description: a long description |
||||
|
* price: required |
||||
|
* tax: if the item doesn't have tax, you can omit this |
||||
|
* tags: a set of unique tag strings for this item |
||||
|
""" |
||||
|
return item |
||||
|
@ -1,9 +1,8 @@ |
|||||
from fastapi import FastAPI |
from fastapi import FastAPI |
||||
from starlette.responses import UJSONResponse |
|
||||
|
|
||||
app = FastAPI() |
app = FastAPI() |
||||
|
|
||||
|
|
||||
@app.get("/items/", content_type=UJSONResponse) |
@app.get("/items/", include_in_schema=False) |
||||
async def read_items(): |
async def read_items(): |
||||
return [{"item_id": "Foo"}] |
return [{"item_id": "Foo"}] |
||||
|
@ -1,18 +1,9 @@ |
|||||
from fastapi import FastAPI |
from fastapi import FastAPI |
||||
from starlette.responses import HTMLResponse |
from starlette.responses import UJSONResponse |
||||
|
|
||||
app = FastAPI() |
app = FastAPI() |
||||
|
|
||||
|
|
||||
@app.get("/items/", content_type=HTMLResponse) |
@app.get("/items/", content_type=UJSONResponse) |
||||
async def read_items(): |
async def read_items(): |
||||
return """ |
return [{"item_id": "Foo"}] |
||||
<html> |
|
||||
<head> |
|
||||
<title>Some HTML in here</title> |
|
||||
</head> |
|
||||
<body> |
|
||||
<h1>Look ma! HTML!</h1> |
|
||||
</body> |
|
||||
</html> |
|
||||
""" |
|
||||
|
@ -1,27 +1,18 @@ |
|||||
from fastapi import Depends, FastAPI |
from fastapi import FastAPI |
||||
from pydantic import BaseModel |
from starlette.responses import HTMLResponse |
||||
|
|
||||
app = FastAPI() |
app = FastAPI() |
||||
|
|
||||
|
|
||||
fake_items_db = [{"item_name": "Foo"}, {"item_name": "Bar"}, {"item_name": "Baz"}] |
@app.get("/items/", content_type=HTMLResponse) |
||||
|
async def read_items(): |
||||
|
return """ |
||||
class CommonQueryParams(BaseModel): |
<html> |
||||
q: str = None |
<head> |
||||
skip: int = None |
<title>Some HTML in here</title> |
||||
limit: int = None |
</head> |
||||
|
<body> |
||||
|
<h1>Look ma! HTML!</h1> |
||||
async def common_parameters(q: str = None, skip: int = 0, limit: int = 100): |
</body> |
||||
return CommonQueryParams(q=q, skip=skip, limit=limit) |
</html> |
||||
|
""" |
||||
|
|
||||
@app.get("/items/") |
|
||||
async def read_items(commons: CommonQueryParams = Depends(common_parameters)): |
|
||||
response = {} |
|
||||
if commons.q: |
|
||||
response.update({"q": commons.q}) |
|
||||
items = fake_items_db[commons.skip : commons.limit] |
|
||||
response.update({"items": items}) |
|
||||
return response |
|
||||
|
@ -1,34 +1,27 @@ |
|||||
from typing import List |
from fastapi import Depends, FastAPI |
||||
|
|
||||
from fastapi import Cookie, Depends, FastAPI |
|
||||
from pydantic import BaseModel |
from pydantic import BaseModel |
||||
|
|
||||
app = FastAPI() |
app = FastAPI() |
||||
|
|
||||
|
|
||||
class InterestsTracker(BaseModel): |
fake_items_db = [{"item_name": "Foo"}, {"item_name": "Bar"}, {"item_name": "Baz"}] |
||||
track_code: str |
|
||||
interests: List[str] |
|
||||
|
|
||||
|
|
||||
fake_tracked_users_db = { |
class CommonQueryParams(BaseModel): |
||||
"Foo": {"track_code": "Foo", "interests": ["sports", "movies"]}, |
q: str = None |
||||
"Bar": {"track_code": "Bar", "interests": ["food", "shows"]}, |
skip: int = None |
||||
"Baz": {"track_code": "Baz", "interests": ["gaming", "virtual reality"]}, |
limit: int = None |
||||
} |
|
||||
|
|
||||
|
|
||||
async def get_tracked_interests(track_code: str = Cookie(None)): |
async def common_parameters(q: str = None, skip: int = 0, limit: int = 100): |
||||
if track_code in fake_tracked_users_db: |
return CommonQueryParams(q=q, skip=skip, limit=limit) |
||||
track_dict = fake_tracked_users_db[track_code] |
|
||||
track = InterestsTracker(**track_dict) |
|
||||
return track |
|
||||
return None |
|
||||
|
|
||||
|
|
||||
@app.get("/interests/") |
@app.get("/items/") |
||||
async def read_interests( |
async def read_items(commons: CommonQueryParams = Depends(common_parameters)): |
||||
tracked_interests: InterestsTracker = Depends(get_tracked_interests) |
response = {} |
||||
): |
if commons.q: |
||||
response = {"interests": tracked_interests.interests} |
response.update({"q": commons.q}) |
||||
|
items = fake_items_db[commons.skip : commons.limit] |
||||
|
response.update({"items": items}) |
||||
return response |
return response |
||||
|
@ -1,52 +1,49 @@ |
|||||
from fastapi import FastAPI |
from random import choice |
||||
|
from typing import List |
||||
|
|
||||
from sqlalchemy import Boolean, Column, Integer, String, create_engine |
from fastapi import Cookie, Depends, FastAPI |
||||
from sqlalchemy.ext.declarative import declarative_base, declared_attr |
from pydantic import BaseModel |
||||
from sqlalchemy.orm import scoped_session, sessionmaker |
|
||||
|
|
||||
# SQLAlchemy specific code, as with any other app |
|
||||
|
|
||||
|
|
||||
SQLALCHEMY_DATABASE_URI = "postgresql://user:password@postgresserver/db" |
|
||||
|
|
||||
# By creating this a CustomBase class and inheriting from it, your models will have |
|
||||
# automatic __tablename__ attributes. So you don't have to declare them. |
|
||||
# So, your models will behave very similarly to, for example, Flask-SQLAlchemy |
|
||||
|
|
||||
|
|
||||
class CustomBase: |
|
||||
# Generate __tablename__ automatically |
|
||||
@declared_attr |
|
||||
def __tablename__(cls): |
|
||||
return cls.__name__.lower() |
|
||||
|
|
||||
|
app = FastAPI() |
||||
|
|
||||
Base = declarative_base(cls=CustomBase) |
|
||||
|
|
||||
|
class InterestsTracker(BaseModel): |
||||
|
track_code: str |
||||
|
interests: List[str] |
||||
|
|
||||
class User(Base): |
|
||||
# Own properties |
|
||||
id = Column(Integer, primary_key=True, index=True) |
|
||||
email = Column(String, unique=True, index=True) |
|
||||
hashed_password = Column(String) |
|
||||
is_active = Column(Boolean(), default=True) |
|
||||
|
|
||||
|
fake_tracked_users_db = { |
||||
|
"Foo": {"track_code": "Foo", "interests": ["sports", "movies"]}, |
||||
|
"Bar": {"track_code": "Bar", "interests": ["food", "shows"]}, |
||||
|
"Baz": {"track_code": "Baz", "interests": ["gaming", "virtual reality"]}, |
||||
|
} |
||||
|
|
||||
engine = create_engine(SQLALCHEMY_DATABASE_URI, convert_unicode=True) |
|
||||
db_session = scoped_session( |
|
||||
sessionmaker(autocommit=False, autoflush=False, bind=engine) |
|
||||
) |
|
||||
|
|
||||
|
async def get_tracked_interests(track_code: str = Cookie(None)): |
||||
|
if track_code in fake_tracked_users_db: |
||||
|
track_dict = fake_tracked_users_db[track_code] |
||||
|
track = InterestsTracker(**track_dict) |
||||
|
return track |
||||
|
return None |
||||
|
|
||||
def get_user(username, db_session): |
|
||||
return db_session.query(User).filter(User.id == username).first() |
|
||||
|
|
||||
|
class ComplexTracker: |
||||
|
def __init__(self, tracker: InterestsTracker = Depends(get_tracked_interests)): |
||||
|
self.tracker = tracker |
||||
|
|
||||
# FastAPI specific code |
def random_interest(self): |
||||
app = FastAPI() |
""" |
||||
|
Get a random interest from the tracked ones for the current user. |
||||
|
If the user doesn't have tracked interests, return a random one from the ones available. |
||||
|
""" |
||||
|
if self.tracker.interests: |
||||
|
return choice(self.tracker.interests) |
||||
|
return choice( |
||||
|
["sports", "movies", "food", "shows", "gaming", "virtual reality"] |
||||
|
) |
||||
|
|
||||
|
|
||||
@app.get("/users/{username}") |
@app.get("/suggested-category") |
||||
def read_user(username: str): |
async def read_suggested_category(tracker: ComplexTracker = Depends(None)): |
||||
user = get_user(username, db_session) |
response = {"category": tracker.random_interest()} |
||||
return user |
return response |
||||
|
Loading…
Reference in new issue