Spaces:
Sleeping
Sleeping
File size: 9,369 Bytes
71fdb6d 9995fd4 71fdb6d 9995fd4 71fdb6d 9995fd4 8ac4c8f 71fdb6d 9995fd4 71fdb6d 9995fd4 8ac4c8f 9995fd4 9bc9256 b39667b 9995fd4 71fdb6d 9995fd4 71fdb6d 9995fd4 71fdb6d 9995fd4 71fdb6d 9995fd4 b39667b 9995fd4 71fdb6d 9995fd4 71fdb6d 8ac4c8f 71fdb6d 9995fd4 71fdb6d 9995fd4 71fdb6d 9995fd4 b39667b 9995fd4 b39667b 9995fd4 8ac4c8f 9995fd4 8ac4c8f 9995fd4 71fdb6d 9995fd4 71fdb6d 9995fd4 71fdb6d 9995fd4 b39667b 9995fd4 71fdb6d |
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 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 |
"""
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()
|