📁 project/
├── 📁 src/
│ ├── 📄 main.py
│ ├── 📄 config.py
│ └── 📁 modules/
│ ├── 📄 auth.py
│ ├── 📄 database.py
│ └── 📄 api.py
├── 📁 tests/
│ └── 📄 test_main.py
└── 📄 requirements.txt
import os
import sys
from typing import Dict, List, Optional
from datetime import datetime, timedelta
import json
import logging
class ConfigManager:
def __init__(self, config_path: str = "config.json"):
self.config_path = config_path
self.config = self._load_config()
def _load_config(self) -> Dict:
try:
with open(self.config_path, 'r') as f:
return json.load(f)
except FileNotFoundError:
return self._get_default_config()
def _get_default_config(self) -> Dict:
return {
"database": {
"host": "localhost",
"port": 5432,
"name": "app_db"
},
"api": {
"host": "0.0.0.0",
"port": 8000,
"debug": True
},
"logging": {
"level": "INFO",
"format": "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
}
}
class DatabaseConnection:
def __init__(self, config: Dict):
self.config = config
self.connection = None
self.cursor = None
def connect(self) -> bool:
try:
import psycopg2
self.connection = psycopg2.connect(
host=self.config["host"],
port=self.config["port"],
database=self.config["name"]
)
self.cursor = self.connection.cursor()
logging.info("Database connection established")
return True
except Exception as e:
logging.error(f"Database connection failed: {e}")
return False
def execute_query(self, query: str, params: tuple = None) -> List:
try:
self.cursor.execute(query, params)
return self.cursor.fetchall()
except Exception as e:
logging.error(f"Query execution failed: {e}")
return []
def close(self):
if self.cursor:
self.cursor.close()
if self.connection:
self.connection.close()
class AuthenticationService:
def __init__(self, db: DatabaseConnection):
self.db = db
self.secret_key = os.environ.get("SECRET_KEY", "default-secret")
def authenticate_user(self, username: str, password: str) -> Optional[str]:
query = "SELECT id, password_hash FROM users WHERE username = %s"
result = self.db.execute_query(query, (username,))
if result and self._verify_password(password, result[0][1]):
return self._generate_token(result[0][0])
return None
def _verify_password(self, password: str, hash_value: str) -> bool:
import bcrypt
return bcrypt.checkpw(password.encode('utf-8'), hash_value.encode('utf-8'))
def _generate_token(self, user_id: int) -> str:
import jwt
payload = {
'user_id': user_id,
'exp': datetime.utcnow() + timedelta(hours=24)
}
return jwt.encode(payload, self.secret_key, algorithm='HS256')
def _get_profile(self, data):
# This would typically extract user_id from JWT token
user_id = data.get("user_id", 1) # Simplified for demo
query = "SELECT username, email, created_at FROM users WHERE id = %s"
result = self.db.execute_query(query, (user_id,))
if result:
user_data = result[0]
return {
"username": user_data[0],
"email": user_data[1],
"created_at": user_data[2].isoformat()
}
else:
return {"error": "User not found"}, 404
def _update_profile(self, data):
user_id = data.get("user_id", 1)
email = data.get("email")
if not email:
return {"error": "Email required"}, 400
query = "UPDATE users SET email = %s, updated_at = %s WHERE id = %s"
self.db.execute_query(query, (email, datetime.utcnow(), user_id))
return {"message": "Profile updated successfully"}
def _auth_middleware(self, data):
# Simplified authentication middleware
token = data.get("authorization", "").replace("Bearer ", "")
if token:
try:
import jwt
payload = jwt.decode(token, self.auth_service.secret_key, algorithms=['HS256'])
data["user_id"] = payload["user_id"]
except:
pass
return data
def _logging_middleware(self, data):
self.logger.info(f"Request processed: {data}")
return data
def start(self):
self.logger.info("Starting application...")
if not self.db.connect():
self.logger.error("Failed to connect to database")
return False
self.logger.info("Application started successfully")
return True
def stop(self):
self.db.close()
self.logger.info("Application stopped")
def main():
app = Application()
if app.start():
try:
# Simulate some API calls
print("Application running...")
print("Available endpoints:")
print("- GET /health")
print("- POST /auth/login")
print("- POST /auth/register")
print("- GET /users/profile")
print("- PUT /users/profile")
except KeyboardInterrupt:
print("\nShutting down...")
finally:
app.stop()
else:
print("Failed to start application")
if __name__ == "__main__":
main()
