article

Friday, March 24, 2023

FastAPI jwt token authentication with sqlite

FastAPI jwt token authentication with sqlite

Install Fastapi

https://github.com/tiangolo/fastapi

pip install fastapi
C:\fastAPI\fastapi-jwt>pip install fastapi
C:\fastAPI\fastapi-jwt>pip install "uvicorn[standard]"


Create main.py
fastapi-jwt/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
#fastapi-jwt/main.py
from typing import List
import fastapi as _fastapi
import fastapi.security as _security
 
import sqlalchemy.orm as _orm
 
import services as _services, schemas as _schemas
 
app = _fastapi.FastAPI()
 
@app.post("/api/users")
async def create_user(
    user: _schemas.UserCreate, db: _orm.Session = _fastapi.Depends(_services.get_db)
):
    db_user = await _services.get_user_by_email(user.email, db)
    if db_user:
        raise _fastapi.HTTPException(status_code=400, detail="Email already in use")
 
    user = await _services.create_user(user, db)
 
    return await _services.create_token(user)
 
 
@app.post("/api/token")
async def generate_token(
    form_data: _security.OAuth2PasswordRequestForm = _fastapi.Depends(),
    db: _orm.Session = _fastapi.Depends(_services.get_db),
):
    user = await _services.authenticate_user(form_data.username, form_data.password, db)
 
    if not user:
        raise _fastapi.HTTPException(status_code=401, detail="Invalid Credentials")
 
    return await _services.create_token(user)
 
 
@app.get("/api/users/myprofile", response_model=_schemas.User)
async def get_user(user: _schemas.User = _fastapi.Depends(_services.get_current_user)):
    return user
install python-jwt
https://pypi.org/project/python-jwt/

pip install python_jwt
C:\fastAPI\react>pip install python_jwt

fastapi-jwt/services.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
#fastapi-jwt/services.py
import fastapi as _fastapi
import fastapi.security as _security
import jwt as _jwt #pip install python_jwt https://pypi.org/project/python-jwt/
import datetime as _dt
import sqlalchemy.orm as _orm
import passlib.hash as _hash
 
import database as _database, models as _models, schemas as _schemas
 
oauth2schema = _security.OAuth2PasswordBearer(tokenUrl="/api/token")
 
JWT_SECRET = "cairocodersednalan"
 
def create_database():
    return _database.Base.metadata.create_all(bind=_database.engine)
 
def get_db():
    db = _database.SessionLocal()
    try:
        yield db
    finally:
        db.close()
 
async def get_user_by_email(email: str, db: _orm.Session):
    return db.query(_models.User).filter(_models.User.email == email).first()
 
async def create_user(user: _schemas.UserCreate, db: _orm.Session):
    user_obj = _models.User(
        email=user.email, hashed_password=_hash.bcrypt.hash(user.hashed_password)
    )
    db.add(user_obj)
    db.commit()
    db.refresh(user_obj)
    return user_obj
 
async def authenticate_user(email: str, password: str, db: _orm.Session):
    user = await get_user_by_email(db=db, email=email)
 
    if not user:
        return False
    if not user.verify_password(password):
        return False
    return user
 
 
async def create_token(user: _models.User):
    user_obj = _schemas.User.from_orm(user)
 
    token = _jwt.encode(user_obj.dict(), JWT_SECRET)
 
    return dict(access_token=token, token_type="bearer")
 
async def get_current_user(
    db: _orm.Session = _fastapi.Depends(get_db),
    token: str = _fastapi.Depends(oauth2schema),
):
    try:
        payload = _jwt.decode(token, JWT_SECRET, algorithms=["HS256"])
        user = db.query(_models.User).get(payload["id"])
    except:
        raise _fastapi.HTTPException(
            status_code=401, detail="Invalid Email or Password"
        )
 
    return _schemas.User.from_orm(user)
fastapi-jwt/database.py
?
1
2
3
4
5
6
7
8
9
10
11
12
#fastapi-jwt/database.py
import sqlalchemy as _sql
import sqlalchemy.ext.declarative as _declarative
import sqlalchemy.orm as _orm
 
DATABASE_URL = "sqlite:///./database.db"
 
engine = _sql.create_engine(DATABASE_URL, connect_args={"check_same_thread": False})
 
SessionLocal = _orm.sessionmaker(autocommit=False, autoflush=False, bind=engine)
 
Base = _declarative.declarative_base()
fastapi-jwt/models.py
?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
#fastapi-jwt/models.py
import datetime as _dt
 
import sqlalchemy as _sql
import sqlalchemy.orm as _orm
import passlib.hash as _hash
 
import database as _database
 
class User(_database.Base):
    __tablename__ = "users"
    id = _sql.Column(_sql.Integer, primary_key=True, index=True)
    email = _sql.Column(_sql.String, unique=True, index=True)
    hashed_password = _sql.Column(_sql.String)
 
    def verify_password(self, password: str):
        return _hash.bcrypt.verify(password, self.hashed_password)
fastapi-jwt/schemas.py
?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
#fastapi-jwt/schemas.py
import datetime as _dt
 
import pydantic as _pydantic
 
class _UserBase(_pydantic.BaseModel):
    email: str
 
class UserCreate(_UserBase):
    hashed_password: str
 
    class Config:
        orm_mode = True
 
class User(_UserBase):
    id: int
 
    class Config:
        orm_mode = True
Run C:\fastAPI\fastapi-jwt>uvicorn main:app --reload

http://127.0.0.1:8000/docs
Posted by ednalan at 1:05 AM . Under .

Related Post