🐛 fix(auth.py): import get_session from correct module to fix import error

✨ feat(auth.py): add support for creating user API key and getting user ID from token
🐛 fix(base.py): fix typo in API_KEY_SECRET_KEY variable name
🐛 fix(base.py): fix typo in FIRST_SUPERUSER and FIRST_SUPERUSER_PASSWORD variable names
🐛 fix(base.py): fix indentation in load_settings_from_yaml function
This commit is contained in:
gustavoschaedler 2023-08-15 23:47:50 +01:00
commit 91fcf33506
2 changed files with 26 additions and 4 deletions

View file

@ -7,9 +7,8 @@ from fastapi.security import OAuth2PasswordBearer
from fastapi import Depends, HTTPException, status from fastapi import Depends, HTTPException, status
from datetime import datetime, timedelta, timezone from datetime import datetime, timedelta, timezone
from langflow.services.utils import get_settings_manager from langflow.services.utils import get_settings_manager, get_session
from langflow.services.utils import get_session
from langflow.database.models.user import ( from langflow.database.models.user import (
User, User,
get_user_by_id, get_user_by_id,
@ -125,6 +124,23 @@ def create_user_longterm_token(db: Session = Depends(get_session)) -> dict:
} }
def create_user_api_key(user_id: UUID) -> dict:
access_token = create_token(
data={"sub": str(user_id), "role": "api_key"},
expires_delta=timedelta(days=365 * 2),
)
return {"api_key": access_token}
def get_user_id_from_token(token: str) -> UUID:
try:
user_id = jwt.get_unverified_claims(token)["sub"]
return UUID(user_id)
except (KeyError, JWTError, ValueError):
return UUID(int=0)
def create_user_tokens( def create_user_tokens(
user_id: UUID, db: Session = Depends(get_session), update_last_login: bool = False user_id: UUID, db: Session = Depends(get_session), update_last_login: bool = False
) -> dict: ) -> dict:

View file

@ -42,11 +42,17 @@ class Settings(BaseSettings):
ACCESS_TOKEN_EXPIRE_MINUTES: int = 60 ACCESS_TOKEN_EXPIRE_MINUTES: int = 60
REFRESH_TOKEN_EXPIRE_MINUTES: int = 70 REFRESH_TOKEN_EXPIRE_MINUTES: int = 70
# API Key to execute /process endpoint
API_KEY_SECRET_KEY: Optional[
str
] = "b82818e0ad4ff76615c5721ee21004b07d84cd9b87ba4d9cb42374da134b841a"
API_KEY_ALGORITHM: str = "HS256"
# If AUTO_LOGIN = True # If AUTO_LOGIN = True
# > The application does not request login and logs in automatically as a super user. # > The application does not request login and logs in automatically as a super user.
AUTO_LOGIN: bool = True AUTO_LOGIN: bool = True
FIRST_SUPERUSER: str = "superuser" FIRST_SUPERUSER: str = "langflow"
FIRST_SUPERUSER_PASSWORD: str = "12345" FIRST_SUPERUSER_PASSWORD: str = "langflow"
@validator("DATABASE_URL", pre=True) @validator("DATABASE_URL", pre=True)
def set_database_url(cls, value): def set_database_url(cls, value):