Skip to content
Snippets Groups Projects
Verified Commit 7ec0fc97 authored by hanfi's avatar hanfi
Browse files

specific imports and more comments

parent 3e148b92
No related branches found
No related tags found
No related merge requests found
import datetime from datetime import datetime, timedelta
from uuid import UUID from uuid import UUID
from fastapi import Depends, FastAPI, HTTPException, Request, status from fastapi import Depends, FastAPI, HTTPException, Request, status
...@@ -18,6 +18,7 @@ create_database() ...@@ -18,6 +18,7 @@ create_database()
app = FastAPI() app = FastAPI()
# CORS handling
origins = [settings.customer_url, settings.worker_url] origins = [settings.customer_url, settings.worker_url]
app.add_middleware( app.add_middleware(
CORSMiddleware, CORSMiddleware,
...@@ -26,14 +27,18 @@ app.add_middleware( ...@@ -26,14 +27,18 @@ app.add_middleware(
allow_methods=["*"], allow_methods=["*"],
allow_headers=["*"], allow_headers=["*"],
) )
# Rate Limiting for some endpoints
limiter = Limiter(key_func=get_remote_address) limiter = Limiter(key_func=get_remote_address)
app.state.limiter = limiter app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler) app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)
# Authentication setup
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token") oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
oauth2_tokener = Serializer(settings.signing_key) oauth2_tokener = Serializer(settings.signing_key)
# Dependency # DB Dependency
def get_db(): def get_db():
db = SessionLocal() db = SessionLocal()
try: try:
...@@ -42,6 +47,7 @@ def get_db(): ...@@ -42,6 +47,7 @@ def get_db():
db.close() db.close()
# Routes
@app.post("/item/prepare", response_model=schemas.Item) @app.post("/item/prepare", response_model=schemas.Item)
@limiter.limit("2/minute") @limiter.limit("2/minute")
def add_item( def add_item(
...@@ -72,10 +78,7 @@ def get_item(item_uuid: str, db: Session = Depends(get_db)): ...@@ -72,10 +78,7 @@ def get_item(item_uuid: str, db: Session = Depends(get_db)):
@app.get("/items", response_model=list[schemas.Item]) @app.get("/items", response_model=list[schemas.Item])
def get_items(token: str = Depends(oauth2_scheme), db: Session = Depends(get_db)): def get_items(token: str = Depends(oauth2_scheme), db: Session = Depends(get_db)):
if ( if datetime.fromtimestamp(oauth2_tokener.loads(token)) < datetime.now():
datetime.datetime.fromtimestamp(oauth2_tokener.loads(token))
< datetime.datetime.now()
):
raise HTTPException( raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED, status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid authentication credentials", detail="Invalid authentication credentials",
...@@ -114,10 +117,7 @@ def verify_supporter(form_data: OAuth2PasswordRequestForm = Depends()): ...@@ -114,10 +117,7 @@ def verify_supporter(form_data: OAuth2PasswordRequestForm = Depends()):
raise HTTPException(status_code=400, detail="Incorrect username or password") raise HTTPException(status_code=400, detail="Incorrect username or password")
return { return {
"access_token": oauth2_tokener.dumps( "access_token": oauth2_tokener.dumps(
( (datetime.now() + timedelta(minutes=settings.token_lifetime)).timestamp()
datetime.datetime.now()
+ datetime.timedelta(minutes=settings.token_lifetime)
).timestamp()
), ),
"token_type": "bearer", "token_type": "bearer",
} }
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment