0

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

Cross
  • 497
  • 1
  • 3
  • 13

0 Answers0