Good afternoon. I'm trying to make rest api for my future blog using fast api, and now I'm doing an authentication. But there's a problem. I'm trying to import my 'get session' method, but interpreter shows me that error:
ImportError: cannot import name 'get_session' from partially initialized module 'store' (most likely due to a circular import)
So here's the file where error appears(from store import get_session, get_one_user):
from datetime import timedelta, datetime
from typing import Optional
from fastapi import Depends, HTTPException
from fastapi.security import OAuth2PasswordBearer
from passlib.context import CryptContext
from sqlalchemy.orm import Session
from schemas import UserModel, UserUpdate, Token, TokenData
from jose import JWTError, jwt
from starlette import status
from configs import SECRET_KEY, ACCESS_TOKEN_EXPIRE_MINUTES, ALGORITHM
from store import get_session, get_one_user
pwd_context = CryptContext(schemes=['bcrypt'], deprecated='auto')
oauth2_scheme = OAuth2PasswordBearer(tokenUrl='token')
def verify_password(plain_password, hashed_password):
return pwd_context.verify(plain_password, hashed_password)
def get_password_hash(password):
return pwd_context.hash(password)
def authenticate_user(session: Session, nickname: str, password: str):
user = get_one_user(user=UserUpdate(
nickname=nickname), session=session
)
if not user:
return False
if not verify_password(password, user.password):
return False
return user
def create_access_token(data: dict, expires_delta: Optional[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
def get_current_user(session: Session = Depends(get_session), token: str = Depends(oauth2_scheme)):
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail='Could not validate credentials',
headers={'WWW-Authenticate': 'Bearer'}
)
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
username: str = payload.get('sub')
if username is None:
raise credentials_exception
token_data = TokenData(username=username)
except JWTError:
raise credentials_exception
user = get_one_user(session=session, user=UserUpdate(nickname=token_data.username))
if user is None:
raise credentials_exception
return user
here is the init file of store:
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from store.user_methods import add_user, get_all_users, user_delete, user_update, get_one_user
from store.db_model import User, Post
from store.auth_methods import get_current_user,authenticate_user,create_access_token
database_protocol = "sqlite:///store/database.db"
engine = create_engine(database_protocol, echo=True, connect_args={"check_same_thread": False})
session = sessionmaker(bind=engine)
def get_session():
sess = session()
try:
yield sess
finally:
sess.close()
and here's the handler that calls this method:
@auth_router.get('/user/me', response_model=UserModel)
def read_users_me(current_user: UserModel = Depends(get_current_user)):
return current_user
What am I doing wrong? If you know, please tell me, I'd really apreciate it