File size: 1,859 Bytes
55f4f81 | 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 | import json
from .mongo import users_collection
from bson.objectid import ObjectId
from redis_client import redis_client , CACHE_TTL
ALLOWED_FIELDS = {"custom_instruction", "tone", "verbosity"}
def get_prompt_by_user(user_id: str) -> dict:
cache_key = f"user_prompt:{user_id}"
try:
cached_data = redis_client.get(cache_key)
if cached_data:
return json.loads(cached_data)
user = users_collection.find_one({"_id": ObjectId(user_id)})
if not user:
return {"error": f"No User found for ID {user_id}."}
prompt_data = {
"custom_instruction": user.get("custom_instruction", ""),
"tone": user.get("tone", "Balanced"),
"verbosity": user.get("verbosity", "Medium"),
}
redis_client.setex(cache_key, CACHE_TTL, json.dumps(prompt_data))
return prompt_data
except Exception as e:
return {"error": str(e)}
def update_prompt_for_user(user_id: str, field: str, value: str) -> dict:
try:
if field not in ALLOWED_FIELDS:
return {"error": "Invalid field"}
result = users_collection.update_one(
{"_id": ObjectId(user_id)},
{"$set": {field: value}},
)
if result.matched_count == 0:
return {"error": "User not found"}
user = users_collection.find_one({"_id": ObjectId(user_id)})
updated_prompt_data = {
"custom_instruction": user.get("custom_instruction", ""),
"tone": user.get("tone", "Balanced"),
"verbosity": user.get("verbosity", "Medium"),
}
cache_key = f"user_prompt:{user_id}"
redis_client.setex(cache_key, CACHE_TTL, json.dumps(updated_prompt_data))
return updated_prompt_data
except Exception as e:
return {"error": str(e)} |