-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathmain.py
164 lines (141 loc) · 5.1 KB
/
main.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
import os
from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session
from typing import List
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import HTMLResponse
from passlib.context import CryptContext
from jose import JWTError, jwt
from datetime import datetime, timedelta
from models import Base, Card, SessionLocal, engine
import schemas
# Create all tables
Base.metadata.create_all(bind=engine)
app = FastAPI()
app.add_middleware(
CORSMiddleware,
allow_origins=["*"], # Adjust this to your needs
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# Dependency
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
def add_default_cards(db: Session):
default_cards = [
{"title": "Card 1", "image": "Image 1"},
{"title": "Card 2", "image": "Image 2"},
{"title": "Card 3", "image": "Image 3"},
]
for card_data in default_cards:
card = Card(**card_data)
db.add(card)
db.commit()
@app.on_event("startup")
def startup_event():
db = SessionLocal()
add_default_cards(db)
db.close()
# Password hashing
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
def get_password_hash(password):
return pwd_context.hash(password)
def verify_password(plain_password, hashed_password):
return pwd_context.verify(plain_password, hashed_password)
# JWT settings
SECRET_KEY = "your_secret_key"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30
def create_access_token(data: dict, expires_delta: timedelta = None):
to_encode = data.copy()
if expires_delta:
expire = datetime.utcnow() + expires_delta
else:
expire = datetime.utcnow() + timedelta(minutes=15)
to_encode.update({"exp": expire})
encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
return encoded_jwt
@app.post("/users/", response_model=schemas.UserResponse)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
db_user = db.query(User).filter(User.email == user.email).first()
if db_user:
raise HTTPException(status_code=400, detail="Email already registered")
hashed_password = get_password_hash(user.password)
db_user = User(email=user.email, hashed_password=hashed_password)
db.add(db_user)
db.commit()
db.refresh(db_user)
return db_user
@app.post("/token", response_model=dict)
def login_for_access_token(form_data: schemas.UserLogin, db: Session = Depends(get_db)):
user = db.query(User).filter(User.email == form_data.email).first()
if not user or not verify_password(form_data.password, user.hashed_password):
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Incorrect email or password",
headers={"WWW-Authenticate": "Bearer"},
)
access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
access_token = create_access_token(
data={"sub": user.email}, expires_delta=access_token_expires
)
return {"access_token": access_token, "token_type": "bearer"}
@app.get("/", response_class=HTMLResponse)
async def read_root():
html_content = """
<!DOCTYPE html>
<html>
<head>
<title>Welcome to FastAPI</title>
</head>
<body>
<h1>Welcome to FastAPI</h1>
<p>This is the landing page of your FastAPI application.</p>
</body>
</html>
"""
return HTMLResponse(content=html_content)
@app.post("/cards/", response_model=schemas.Card)
def create_card(card: schemas.CardCreate, db: Session = Depends(get_db)):
db_card = Card(title=card.title, image=card.image)
db.add(db_card)
db.commit()
db.refresh(db_card)
return db_card
@app.get("/cards/", response_model=List[schemas.Card])
def read_cards(skip: int = 0, limit: int = 10, db: Session = Depends(get_db)):
cards = db.query(Card).offset(skip).limit(limit).all()
return cards
@app.get("/cards/{card_id}", response_model=schemas.Card)
def read_card(card_id: int, db: Session = Depends(get_db)):
card = db.query(Card).filter(Card.id == card_id).first()
if card is None:
raise HTTPException(status_code=404, detail="Card not found")
return card
@app.put("/cards/{card_id}", response_model=schemas.Card)
def update_card(card_id: int, card: schemas.CardCreate, db: Session = Depends(get_db)):
db_card = db.query(Card).filter(Card.id == card_id).first()
if db_card is None:
raise HTTPException(status_code=404, detail="Card not found")
db_card.title = card.title
db_card.image = card.image
db.commit()
db.refresh(db_card)
return db_card
@app.delete("/cards/{card_id}", response_model=schemas.Card)
def delete_card(card_id: int, db: Session = Depends(get_db)):
db_card = db.query(Card).filter(Card.id == card_id).first()
if db_card is None:
raise HTTPException(status_code=404, detail="Card not found")
db.delete(db_card)
db.commit()
return db_card
if __name__ == "__main__":
port = int(os.environ.get("PORT", 8000))
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=port)