Not a member of Pastebin yet?
Sign Up,
it unlocks many cool features!
- from typing import List
- from datetime import datetime, timedelta
- #from supertokens_fastapi import get_cors_allowed_headers
- from fastapi import Depends, FastAPI, HTTPException, Form
- from sqlalchemy.orm import Session
- from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
- from jose import JWTError, jwt
- #from . import crud, models, schemas
- import crud
- import models
- import schemas
- #from .database import SessionLocal, engine
- from database import SessionLocal, engine
- #from fastapi.middleware.cors import CORSMiddleware
- from starlette.middleware.cors import CORSMiddleware
- models.Base.metadata.create_all(bind=engine)
- app = FastAPI()
- origins = [
- "http://localhost:3001",
- "http://localhost:3000",
- ]
- SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
- ALGORITHM = "HS256"
- ACCESS_TOKEN_EXPIRE_MINUTES = 30
- app.add_middleware(
- CORSMiddleware,
- allow_origins=origins,
- allow_credentials=True,
- #allow_origin_regex = ["https://.*\.example\.org"],
- allow_methods=["*"],
- allow_headers=["*"],
- expose_headers = []
- )
- def get_db():
- db = SessionLocal()
- try:
- yield db
- finally:
- db.close()
- BOOKS = [
- {
- 'title': 'On the Road',
- 'author': 'Jack Kerouac',
- },
- {
- 'title': 'Harry Potter and the Philosopher\'s Stone',
- 'author': 'J. K. Rowling',
- },
- {
- 'title': 'Green Eggs and Ham',
- 'author': 'Dr. Seuss',
- }
- ]
- @app.get("/")
- async def root():
- return {"message": "Hello World"}
- @app.get("/books")
- async def rat():
- return BOOKS
- #auth
- @app.post("/token", response_model=schemas.Token)
- async def login_for_access_token(username: str = Form(...), password: str = Form(...), db: Session = Depends(get_db)):
- print({username})
- print({ password})
- user = crud.authenticate_user(db, username, password)
- if not user:
- raise HTTPException(
- status_code=status.HTTP_401_UNAUTHORIZED,
- detail="Incorrect username or password",
- headers={"WWW-Authenticate": "Bearer"},
- )
- access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
- access_token = crud.create_access_token(
- data={"sub": user.email}, expires_delta=access_token_expires
- )
- return {"access_token": access_token, "token_type": "bearer"}
- @app.get("/users/me/", response_model=schemas.Token)
- async def read_users_me(current_user: models.User = Depends(crud.get_current_active_user)):
- return current_user
- @app.get("/users/me/items/")
- async def read_own_items(current_user: models.User = Depends(crud.get_current_active_user)):
- return [{"item_id": "Foo", "owner": current_user.email}]
- #endauth
- @app.post("/users/", response_model=schemas.User)
- async def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
- db_user = crud.get_user_by_email(db, email=user.email)
- if db_user:
- raise HTTPException(status_code=400, detail="Email already registered")
- return crud.create_user(db=db, user=user)
- @app.get("/users/", response_model=List[schemas.User])
- async def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
- users = crud.get_users(db, skip=skip, limit=limit)
- return users
- @app.get("/users/{user_id}", response_model=schemas.User)
- async def read_user(user_id: int, db: Session = Depends(get_db)):
- db_user = crud.get_user(db, user_id=user_id)
- if db_user is None:
- raise HTTPException(status_code=404, detail="User not found")
- return db_user
- @app.delete("/users/{user_id}", response_model=schemas.UserDelete)
- async def delete_user(user_id: int, db: Session = Depends(get_db)):
- db_user = crud.delete_user(db, user_id=user_id)
- return db_user
- @app.put("/users/{user_id}", response_model=schemas.User)
- async def put_user(user_id: int, user: schemas.UserUpdate, db: Session = Depends(get_db)):
- return crud.update_user(db=db, user_id=user_id, user=user )
- @app.post("/users/{user_id}/items/", response_model=schemas.Item)
- async def create_item_for_user(
- user_id: int, item: schemas.ItemCreate, db: Session = Depends(get_db)
- ):
- return crud.create_user_item(db=db, item=item, user_id=user_id)
- @app.get("/items/", response_model=List[schemas.Item])
- async def read_items(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
- items = crud.get_items(db, skip=skip, limit=limit)
- return items
Add Comment
Please, Sign In to add comment