--- /dev/null
+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
-from fastapi import FastAPI
-from pydantic import BaseModel
+from fastapi import Body, FastAPI
+from pydantic import BaseModel, Schema
app = FastAPI()
description: str = None
price: float
tax: float = None
- tags: list = []
@app.put("/items/{item_id}")
-async def update_item(*, item_id: int, item: Item):
+async def update_item(
+ *,
+ item_id: int,
+ item: Item = Body(
+ ...,
+ example={
+ "name": "Foo",
+ "description": "A very nice Item",
+ "price": 35.4,
+ "tax": 3.2,
+ },
+ )
+):
results = {"item_id": item_id, "item": item}
return results
-from typing import List
-
from fastapi import FastAPI
from pydantic import BaseModel
description: str = None
price: float
tax: float = None
- tags: List[str] = []
+ tags: list = []
@app.put("/items/{item_id}")
-from typing import Set
+from typing import List
from fastapi import FastAPI
from pydantic import BaseModel
description: str = None
price: float
tax: float = None
- tags: Set[str] = []
+ tags: List[str] = []
@app.put("/items/{item_id}")
app = FastAPI()
-class Image(BaseModel):
- url: str
- name: str
-
-
class Item(BaseModel):
name: str
description: str = None
price: float
tax: float = None
tags: Set[str] = []
- image: Image = None
@app.put("/items/{item_id}")
from fastapi import FastAPI
from pydantic import BaseModel
-from pydantic.types import UrlStr
app = FastAPI()
class Image(BaseModel):
- url: UrlStr
+ url: str
name: str
-from typing import List, Set
+from typing import Set
from fastapi import FastAPI
from pydantic import BaseModel
price: float
tax: float = None
tags: Set[str] = []
- image: List[Image] = None
+ image: Image = None
@app.put("/items/{item_id}")
image: List[Image] = None
-class Offer(BaseModel):
- name: str
- description: str = None
- price: float
- items: List[Item]
-
-
-@app.post("/offers/")
-async def create_offer(*, offer: Offer):
- return offer
+@app.put("/items/{item_id}")
+async def update_item(*, item_id: int, item: Item):
+ results = {"item_id": item_id, "item": item}
+ return results
-from typing import List
+from typing import List, Set
from fastapi import FastAPI
from pydantic import BaseModel
name: str
-@app.post("/images/multiple/")
-async def create_multiple_images(*, images: List[Image]):
- return images
+class Item(BaseModel):
+ name: str
+ description: str = None
+ price: float
+ tax: float = None
+ tags: Set[str] = []
+ image: List[Image] = None
+
+
+class Offer(BaseModel):
+ name: str
+ description: str = None
+ price: float
+ items: List[Item]
+
+
+@app.post("/offers/")
+async def create_offer(*, offer: Offer):
+ return offer
-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.get("/items/")
-async def read_items(*, ads_id: str = Cookie(None)):
- return {"ads_id": ads_id}
+class Image(BaseModel):
+ url: UrlStr
+ name: str
+
+
+@app.post("/images/multiple/")
+async def create_multiple_images(*, images: List[Image]):
+ return images
-from fastapi import FastAPI, Header
+from fastapi import Cookie, FastAPI
app = FastAPI()
@app.get("/items/")
-async def read_items(*, accept_encoding: str = Header(None)):
- return {"Accept-Encoding": accept_encoding}
+async def read_items(*, ads_id: str = Cookie(None)):
+ return {"ads_id": ads_id}
@app.get("/items/")
-async def read_items(*, strange_header: str = Header(None, convert_underscores=False)):
- return {"strange_header": strange_header}
+async def read_items(*, accept_encoding: str = Header(None)):
+ return {"Accept-Encoding": accept_encoding}
-from typing import Set
-
-from fastapi import FastAPI
-from pydantic import BaseModel
+from fastapi import FastAPI, Header
app = FastAPI()
-class Item(BaseModel):
- name: str
- description: str = None
- price: float
- tax: float = None
- tags: Set[str] = []
-
-
-@app.post("/items/", response_model=Item)
-async def create_item(*, item: Item):
- return item
+@app.get("/items/")
+async def read_items(*, strange_header: str = Header(None, convert_underscores=False)):
+ return {"strange_header": strange_header}
+from typing import Set
+
from fastapi import FastAPI
from pydantic import BaseModel
-from pydantic.types import EmailStr
app = FastAPI()
-class UserIn(BaseModel):
- username: str
- password: str
- email: EmailStr
- full_name: str = None
+class Item(BaseModel):
+ name: str
+ description: str = None
+ price: float
+ tax: float = None
+ tags: Set[str] = []
-# Don't do this in production!
-@app.post("/user/", response_model=UserIn)
-async def create_user(*, user: UserIn):
- return user
+@app.post("/items/", response_model=Item)
+async def create_item(*, item: Item):
+ return item
full_name: str = None
-class UserOut(BaseModel):
- username: str
- email: EmailStr
- full_name: str = None
-
-
-@app.post("/user/", response_model=UserOut)
+# Don't do this in production!
+@app.post("/user/", response_model=UserIn)
async def create_user(*, user: UserIn):
return user
full_name: str = None
-class UserInDB(BaseModel):
- username: str
- hashed_password: str
- email: EmailStr
- full_name: str = None
-
-
-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
+async def create_user(*, user: UserIn):
+ return user
app = FastAPI()
-class UserBase(BaseModel):
+class UserIn(BaseModel):
username: str
+ password: str
email: EmailStr
full_name: str = None
-class UserIn(UserBase):
- password: str
-
-
-class UserOut(UserBase):
- pass
+class UserOut(BaseModel):
+ username: str
+ email: EmailStr
+ full_name: str = None
-class UserInDB(UserBase):
+class UserInDB(BaseModel):
+ username: str
hashed_password: str
+ email: EmailStr
+ full_name: str = None
def fake_password_hasher(raw_password: str):
-from fastapi import FastAPI, Form
+from fastapi import FastAPI
+from pydantic import BaseModel
+from pydantic.types import EmailStr
app = FastAPI()
-@app.post("/login/")
-async def login(*, username: str = Form(...), password: str = Form(...)):
- return {"username": username}
+class UserBase(BaseModel):
+ username: str
+ 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
-from fastapi import FastAPI, File
+from fastapi import FastAPI, Form
app = FastAPI()
-@app.post("/files/")
-async def create_file(*, file: bytes = File(...)):
- return {"file_size": len(file)}
+@app.post("/login/")
+async def login(*, username: str = Form(...), password: str = Form(...)):
+ return {"username": username}
-from fastapi import FastAPI, File, Form
+from fastapi import FastAPI, File
app = FastAPI()
@app.post("/files/")
-async def create_file(*, file: bytes = File(...), token: str = Form(...)):
- return {"file_size": len(file), "token": token}
+async def create_file(*, file: bytes = File(...)):
+ return {"file_size": len(file)}
-from typing import Set
-
-from fastapi import FastAPI
-from pydantic import BaseModel
-from starlette.status import HTTP_201_CREATED
+from fastapi import FastAPI, File, Form
app = FastAPI()
-class Item(BaseModel):
- name: str
- description: str = None
- 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
+@app.post("/files/")
+async def create_file(*, file: bytes = File(...), token: str = Form(...)):
+ return {"file_size": len(file), "token": token}
from fastapi import FastAPI
from pydantic import BaseModel
+from starlette.status import HTTP_201_CREATED
app = FastAPI()
tags: Set[str] = []
-@app.post("/items/", response_model=Item, tags=["items"])
+@app.post("/items/", response_model=Item, status_code=HTTP_201_CREATED)
async def create_item(*, item: Item):
return item
tags: Set[str] = []
-@app.post(
- "/items/",
- response_model=Item,
- summary="Create an item",
- description="Create an item with all the information, name, description, price, tax and a set of unique tags",
-)
+@app.post("/items/", response_model=Item, tags=["items"])
async def create_item(*, item: Item):
return item
tags: Set[str] = []
-@app.post("/items/", response_model=Item, summary="Create an item")
+@app.post(
+ "/items/",
+ response_model=Item,
+ summary="Create an item",
+ description="Create an item with all the information, name, description, price, tax and a set of unique tags",
+)
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
tags: Set[str] = []
-@app.post(
- "/items/",
- response_model=Item,
- summary="Create an item",
- response_description="The created item",
-)
+@app.post("/items/", response_model=Item, summary="Create an item")
async def create_item(*, item: Item):
"""
Create an item with all the information:
+from typing import Set
+
from fastapi import FastAPI
+from pydantic import BaseModel
app = FastAPI()
-@app.get("/items/", deprecated=True)
-async def read_items():
- return [{"item_id": "Foo"}]
+class Item(BaseModel):
+ name: str
+ 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
app = FastAPI()
-@app.get("/items/", operation_id="some_specific_id_you_define")
+@app.get("/items/", deprecated=True)
async def read_items():
return [{"item_id": "Foo"}]
app = FastAPI()
-@app.get("/items/", include_in_schema=False)
+@app.get("/items/", operation_id="some_specific_id_you_define")
async def read_items():
return [{"item_id": "Foo"}]
from fastapi import FastAPI
-from starlette.responses import UJSONResponse
app = FastAPI()
-@app.get("/items/", content_type=UJSONResponse)
+@app.get("/items/", include_in_schema=False)
async def read_items():
return [{"item_id": "Foo"}]
from fastapi import FastAPI
-from starlette.responses import HTMLResponse
+from starlette.responses import UJSONResponse
app = FastAPI()
-@app.get("/items/", content_type=HTMLResponse)
+@app.get("/items/", content_type=UJSONResponse)
async def read_items():
- return """
- <html>
- <head>
- <title>Some HTML in here</title>
- </head>
- <body>
- <h1>Look ma! HTML!</h1>
- </body>
- </html>
- """
+ return [{"item_id": "Foo"}]
-from fastapi import Depends, FastAPI
-from pydantic import BaseModel
+from fastapi import FastAPI
+from starlette.responses import HTMLResponse
app = FastAPI()
-fake_items_db = [{"item_name": "Foo"}, {"item_name": "Bar"}, {"item_name": "Baz"}]
-
-
-class CommonQueryParams(BaseModel):
- q: str = None
- skip: int = None
- limit: int = None
-
-
-async def common_parameters(q: str = None, skip: int = 0, limit: int = 100):
- return CommonQueryParams(q=q, skip=skip, limit=limit)
-
-
-@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
+@app.get("/items/", content_type=HTMLResponse)
+async def read_items():
+ return """
+ <html>
+ <head>
+ <title>Some HTML in here</title>
+ </head>
+ <body>
+ <h1>Look ma! HTML!</h1>
+ </body>
+ </html>
+ """
-from typing import List
-
-from fastapi import Cookie, Depends, FastAPI
+from fastapi import Depends, FastAPI
from pydantic import BaseModel
app = FastAPI()
-class InterestsTracker(BaseModel):
- track_code: str
- interests: List[str]
+fake_items_db = [{"item_name": "Foo"}, {"item_name": "Bar"}, {"item_name": "Baz"}]
-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"]},
-}
+class CommonQueryParams(BaseModel):
+ q: str = None
+ skip: int = None
+ limit: int = None
-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
+async def common_parameters(q: str = None, skip: int = 0, limit: int = 100):
+ return CommonQueryParams(q=q, skip=skip, limit=limit)
-@app.get("/interests/")
-async def read_interests(
- tracked_interests: InterestsTracker = Depends(get_tracked_interests)
-):
- response = {"interests": tracked_interests.interests}
+@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
-from random import choice
from typing import List
from fastapi import Cookie, Depends, FastAPI
return None
-class ComplexTracker:
- def __init__(self, tracker: InterestsTracker = Depends(get_tracked_interests)):
- self.tracker = tracker
-
- def random_interest(self):
- """
- 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("/suggested-category")
-async def read_suggested_category(tracker: ComplexTracker = Depends(None)):
- response = {"category": tracker.random_interest()}
+@app.get("/interests/")
+async def read_interests(
+ tracked_interests: InterestsTracker = Depends(get_tracked_interests)
+):
+ response = {"interests": tracked_interests.interests}
return response
-from fastapi import FastAPI
+from random import choice
+from typing import List
-from sqlalchemy import Boolean, Column, Integer, String, create_engine
-from sqlalchemy.ext.declarative import declarative_base, declared_attr
-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()
+from fastapi import Cookie, Depends, FastAPI
+from pydantic import BaseModel
+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
-app = FastAPI()
+ def random_interest(self):
+ """
+ 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}")
-def read_user(username: str):
- user = get_user(username, db_session)
- return user
+@app.get("/suggested-category")
+async def read_suggested_category(tracker: ComplexTracker = Depends(None)):
+ response = {"category": tracker.random_interest()}
+ return response
-from typing import Optional
-
from fastapi import FastAPI
-from pydantic import BaseModel
-from app.models.config import USERPROFILE_DOC_TYPE
-from couchbase import LOCKMODE_WAIT
-from couchbase.bucket import Bucket
-from couchbase.cluster import Cluster, PasswordAuthenticator
+from sqlalchemy import Boolean, Column, Integer, String, create_engine
+from sqlalchemy.ext.declarative import declarative_base, declared_attr
+from sqlalchemy.orm import scoped_session, sessionmaker
+# SQLAlchemy specific code, as with any other app
-def get_bucket():
- cluster = Cluster("couchbase://couchbasehost:8091")
- authenticator = PasswordAuthenticator("username", "password")
- cluster.authenticate(authenticator)
- bucket: Bucket = cluster.open_bucket("bucket_name", lockmode=LOCKMODE_WAIT)
- return bucket
+SQLALCHEMY_DATABASE_URI = "postgresql://user:password@postgresserver/db"
-class User(BaseModel):
- username: str
- email: Optional[str] = None
- full_name: Optional[str] = None
- disabled: Optional[bool] = None
+# 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 UserInDB(User):
- type: str = USERPROFILE_DOC_TYPE
- hashed_password: str
+class CustomBase:
+ # Generate __tablename__ automatically
+ @declared_attr
+ def __tablename__(cls):
+ return cls.__name__.lower()
- class Meta:
- key: Optional[str] = None
+Base = declarative_base(cls=CustomBase)
-def get_user_doc_id(username):
- return f"userprofile::{username}"
+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)
-def get_user(bucket: Bucket, username: str):
- doc_id = get_user_doc_id(username)
- result = bucket.get(doc_id, quiet=True)
- if not result.value:
- return None
- user = UserInDB(**result.value)
- user.Meta.key = result.key
- return user
+
+engine = create_engine(SQLALCHEMY_DATABASE_URI, convert_unicode=True)
+db_session = scoped_session(
+ sessionmaker(autocommit=False, autoflush=False, bind=engine)
+)
+
+
+def get_user(username, db_session):
+ return db_session.query(User).filter(User.id == username).first()
# FastAPI specific code
@app.get("/users/{username}")
def read_user(username: str):
- bucket = get_bucket()
- user = get_user(bucket=bucket, username=username)
+ user = get_user(username, db_session)
return user
projects / thirdparty / fastapi / fastapi.git / commitdiff
