📦 chore(api_key/crud.py): add CRUD functions for managing API keys in the database

This commit adds the following functions to the `api_key/crud.py` file:
- `get_api_keys`: Retrieves a list of API keys associated with a specific user ID from the database.
- `create_api_key`: Generates a random API key, hashes it, and creates a new `ApiKey` object in the database.
- `delete_api_key`: Deletes an API key from the database based on its ID.

These functions provide the necessary functionality for managing API keys in the application's database.
This commit is contained in:
Gabriel Luiz Freitas Almeida 2023-08-25 12:04:20 -03:00
commit e42686738c

View file

@ -0,0 +1,44 @@
import secrets
from uuid import UUID
from typing import List
from langflow.services.auth.utils import get_password_hash
from sqlmodel import Session, select
from langflow.services.database.models.api_key import (
ApiKey,
ApiKeyCreate,
UnmaskedApiKeyRead,
ApiKeyRead,
)
def get_api_keys(session: Session, user_id: str) -> List[UnmaskedApiKeyRead]:
query = select(ApiKey).where(ApiKey.user_id == user_id)
api_keys = session.exec(query).all()
return [ApiKeyRead.from_orm(api_key) for api_key in api_keys]
def create_api_key(
session: Session, api_key_create: ApiKeyCreate, user_id: str
) -> UnmaskedApiKeyRead:
# Generate a random API key with 32 bytes of randomness
generated_api_key = secrets.token_urlsafe(32)
# hash the API key
hashed_api_key = get_password_hash(generated_api_key)
# Use the generated key to create the ApiKey object
api_key = ApiKey(api_key=hashed_api_key, **api_key_create.dict(), user_id=user_id)
session.add(api_key)
session.commit()
session.refresh(api_key)
unmasked = UnmaskedApiKeyRead.from_orm(api_key)
unmasked.api_key = generated_api_key
return unmasked
def delete_api_key(session: Session, api_key_id: UUID) -> None:
api_key = session.get(ApiKey, api_key_id)
if api_key is None:
raise ValueError("API Key not found")
session.delete(api_key)
session.commit()