""" Service for storing and retrieving profile data using SQLite """ import sqlite3 from models import Profile, Skill, Project, Education, SocialMedia from config import get_settings import json import logging from typing import Dict, Any, Optional, List import requests settings = get_settings() logger = logging.getLogger(__name__) class StorageService: """Service for storing and retrieving profile data using SQLite""" def __init__(self): self.db_path = settings.SQLITE_DB_PATH self.external_api_url = settings.EXTERNAL_API_URL self._create_table() def _create_table(self): """Create the profiles table if it doesn't exist""" try: conn = sqlite3.connect(self.db_path) cursor = conn.cursor() cursor.execute(""" CREATE TABLE IF NOT EXISTS profiles ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL, title TEXT NOT NULL, email TEXT NOT NULL, bio TEXT NOT NULL, tagline TEXT, social TEXT, profileImg TEXT, projects TEXT, skills TEXT, educations TEXT, experiences TEXT ) """) conn.commit() conn.close() logger.info("Profiles table created or already exists") except Exception as e: logger.error(f"Error creating table: {e}") raise def profile_to_dict(self, profile: Profile) -> Dict[str, Any]: """Convert Profile object to dictionary for SQLite storage""" return { "name": profile.name, "title": profile.title, "email": profile.email, "bio": profile.bio, "tagline": profile.tagline if profile.tagline else None, "social": json.dumps({ "linkedin": profile.social.linkedin if profile.social else None, "github": profile.social.github if profile.social else None, "instagram": profile.social.instagram if profile.social else None }), "profileImg": profile.profileImg, "projects": json.dumps([ { "title": project.title, "description": project.description, "techStack": project.techStack, "githubUrl": project.githubUrl, "demoUrl": project.demoUrl } for project in profile.projects ]), "skills": json.dumps([ { "name": skill.name, "category": skill.category.value if skill.category else None, "img": skill.img } for skill in profile.skills ]), "experiences": json.dumps([ { "company": exp.company, "position": exp.position, "startDate": exp.startDate, "endDate": exp.endDate, "description": exp.description } for exp in profile.experiences ]), "educations": json.dumps([ { "school": edu.school, "degree": edu.degree, "fieldOfStudy": edu.fieldOfStudy, "startDate": edu.startDate, "endDate": edu.endDate } for edu in profile.educations ]) } def send_to_external_api(self, profile_data: Dict[str, Any], profile_id: str) -> Dict[str, Any]: """ Send profile data to the external API Args: profile_data: Dictionary containing profile data profile_id: ID of the stored profile Returns: Response from the external API or error details """ try: if not self.external_api_url: logger.warning("EXTERNAL_API_URL is not configured, skipping external API sync") return {"success": False, "reason": "External API URL not configured"} # Add the ID to the profile data profile_data["id"] = profile_id # Send the data to the external API response = requests.post( f"{self.external_api_url}/profiles", json=profile_data, headers={"Content-Type": "application/json"} ) # Check if the request was successful if response.status_code in (200, 201): logger.info(f"Profile successfully sent to external API") return { "success": True, "external_id": response.json().get("id", None), "external_url": f"{self.external_api_url}/profiles/{profile_id}" } else: logger.error(f"Failed to send profile to external API: {response.status_code} - {response.text}") return { "success": False, "status_code": response.status_code, "message": response.text } except Exception as e: logger.error(f"Error sending profile to external API: {e}") return { "success": False, "exception": str(e) } def store_profile(self, profile: Profile, error_handler=None) -> str: """ Store profile data in SQLite Args: profile: The Profile object to store error_handler: Optional function to handle errors (useful for framework-specific error handling) Returns: String ID of the stored profile """ profile_dict = self.profile_to_dict(profile) try: conn = sqlite3.connect(self.db_path) cursor = conn.cursor() cursor.execute(""" INSERT INTO profiles (name, title, email, bio, tagline, social, profileImg, projects, skills, educations, experiences) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( profile_dict["name"], profile_dict["title"], profile_dict["email"], profile_dict["bio"], profile_dict["tagline"], profile_dict["social"], profile_dict["profileImg"], profile_dict["projects"], profile_dict["skills"], profile_dict["educations"], profile_dict["experiences"] )) conn.commit() profile_id = str(cursor.lastrowid) conn.close() logger.info(f"Profile saved successfully with ID: {profile_id}") # Send to external API external_api_result = self.send_to_external_api(profile_dict, profile_id) # Store the external API result in the session state for later use import streamlit as st if "st" in globals() and hasattr(st, "session_state"): st.session_state.external_api_result = external_api_result return profile_id except Exception as e: logger.error(f"SQLite error: {e}") if error_handler: error_handler(f"Error connecting to SQLite: {str(e)}") return None def get_profile(self, profile_id: int) -> Optional[Dict[str, Any]]: """ Retrieve a profile from SQLite by its ID Args: profile_id: The ID of the profile to retrieve Returns: A dictionary representing the profile, or None if not found """ try: conn = sqlite3.connect(self.db_path) cursor = conn.cursor() cursor.execute("SELECT * FROM profiles WHERE id = ?", (profile_id,)) row = cursor.fetchone() conn.close() if row: profile = { "id": row[0], "name": row[1], "title": row[2], "email": row[3], "bio": row[4], "tagline": row[5], "social": json.loads(row[6]) if row[6] else None, "profileImg": row[7], "projects": json.loads(row[8]) if row[8] else [], "skills": json.loads(row[9]) if row[9] else [], "educations": json.loads(row[10]) if row[10] else [], "experiences": json.loads(row[11]) if row[11] else [] } logger.debug(f"Retrieved profile: {profile_id}") return profile else: logger.warning(f"Profile not found: {profile_id}") return None except Exception as e: logger.error(f"Error retrieving profile {profile_id}: {e}") return None # Create a global instance storage_service = StorageService()