def calculate_total(items):
total = 0
for item in items:
if item["type"] == "A":
total += item["price"] * 0.9 # 10% discount
elif item["type"] == "B":
total += item["price"] * 0.8 # 20% discount
else:
total += item["price"]
return total

items = [
{"type": "A", "price": 100},
{"type": "B", "price": 200},
{"type": "C", "price": 300},
]

def calculate_total(items):
discounts = {
"A": 0.9,
"B": 0.8,
}
total = 0
for item in items:
discount = discounts.get(item["type"], 1.0) # Default to 1.0 (no discount)
total += item["price"] * discount
return total

items = [
{"type": "A", "price": 100},
{"type": "B", "price": 200},
{"type": "C", "price": 300},
]

total = calculate_total(items)
print(f"Total: {total}")

# reusable_module.py

def greet(name):
return f"Hello, {name}!"

def add(x, y):
return x + y

# In another Python file:
import reusable_module

message = reusable_module.greet("Alice")
print(message) # Output: Hello, Alice!

sum_result = reusable_module.add(5, 3)
print(sum_result) # Output: 8

import pandas as pd
from sklearn.model_selection import train_test_split
from sklearn.linear_model import LogisticRegression
from sklearn.metrics import accuracy_score

# 1. Load the data (replace with your data loading)
data = pd.read_csv("customer_data.csv")

# 2. Data Preparation (simplified)
# Assuming 'churn' is the target variable (0 or 1)
# and other columns are features
X = data.drop("churn", axis=1)
y = data["churn"]

# 3. Split data into training and testing sets
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)

# 4. Model Training (Logistic Regression)
model = LogisticRegression()
model.fit(X_train, y_train)

# 5. Model Evaluation
y_pred = model.predict(X_test)
accuracy = accuracy_score(y_test, y_pred)
print(f"Accuracy: {accuracy}")

#!/bin/bash

# List of servers to patch (could be read from a file or inventory system)
SERVERS=( "server01.example.com" "server02.example.com" "server03.example.com" ... "server100.example.com" )

# Log file
LOG_FILE="/var/log/patching_$(date +%Y%m%d_%H%M%S).log"

# SSH options (ensure key-based authentication is set up!)
SSH_OPTS="-o StrictHostKeyChecking=no -o ConnectTimeout=10"

log_message() {
echo "$(date +'%Y-%m-%d %H:%M:%S') - $1" | tee -a "$LOG_FILE"
}

log_message "Starting patching run..."

for server in "${SERVERS[@]}"; do
log_message "Processing server: $server"

# Check connectivity
ping -c 1 "$server" > /dev/null 2>&1
if [ $? -ne 0 ]; then
log_message "ERROR: Cannot reach $server. Skipping."
continue
fi

log_message "Applying updates on $server..."
# Use ssh to run commands remotely. Assumes passwordless sudo is configured or run as root.
# The specific update command depends on the OS (apt, yum, dnf, etc.)
ssh $SSH_OPTS "$server" "sudo apt-get update && sudo apt-get upgrade -y" >> "$LOG_FILE" 2>&1
if [ $? -ne 0 ]; then
log_message "ERROR: Update failed on $server."
# Decide whether to continue or stop the whole process
# continue
else
log_message "Updates applied successfully on $server."
log_message "Rebooting $server..."
ssh $SSH_OPTS "$server" "sudo reboot" >> "$LOG_FILE" 2>&1
# Add a wait period and check if server comes back online (more complex logic needed here)
sleep 60 # Simple wait
log_message "$server reboot initiated."
fi

# Optional: Add delay between servers
# sleep 30
done

log_message "Patching run finished."
exit 0

import re
import os
import glob
from collections import Counter
import csv
from datetime import datetime

# Configuration
LOG_DIR = "/mnt/central_logs/webserver/" # Directory where logs are collected
LOG_PATTERN = "access.log-*" # Pattern to match log files
OUTPUT_CSV = f"/var/reports/ip_summary_{datetime.now():%Y%m%d_%H%M}.csv"
IP_REGEX = r'^(\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3})' # Simple regex to get IP at start of line

def process_logs(log_directory, file_pattern, ip_regex):
"""
Processes log files matching a pattern in a directory.
Returns a Counter object with IP address frequencies.
"""
ip_counter = Counter()
log_files = glob.glob(os.path.join(log_directory, file_pattern))
print(f"Found {len(log_files)} log files to process.")

if not log_files:
print("No log files found matching the pattern.")
return ip_counter

for log_file in log_files:
print(f"Processing {log_file}...")
try:
with open(log_file, 'r', encoding='utf-8', errors='ignore') as f:
for line in f:
match = re.match(ip_regex, line)
if match:
ip_address = match.group(1)
ip_counter[ip_address] += 1
except FileNotFoundError:
print(f"Warning: File {log_file} not found (might have been rotated). Skipping.")
except Exception as e:
print(f"Error processing file {log_file}: {e}")
# Decide whether to continue or stop

return ip_counter

def write_report(ip_counts, output_file):
"""Writes the IP counts to a CSV file."""
print(f"Writing report to {output_file}...")
try:
with open(output_file, 'w', newline='') as csvfile:
fieldnames = ['IP_Address', 'Request_Count']
writer = csv.DictWriter(csvfile, fieldnames=fieldnames)

writer.writeheader()
# Sort by count descending for the report
for ip, count in ip_counts.most_common():
writer.writerow({'IP_Address': ip, 'Request_Count': count})
print("Report written successfully.")
except Exception as e:
print(f"Error writing report file {output_file}: {e}")

if __name__ == "__main__":
print("Starting log analysis...")
start_time = datetime.now()

ip_frequencies = process_logs(LOG_DIR, LOG_PATTERN, IP_REGEX)

if ip_frequencies:
write_report(ip_frequencies, OUTPUT_CSV)
else:
print("No data processed.")

end_time = datetime.now()
print(f"Log analysis finished. Total time: {end_time - start_time}")

# Conceptual Python script - requires specific libraries/SDKs for each system

import requests # For REST APIs
import subprocess # For CLI tools
import json
import logging

# --- Configuration ---
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

# User details (would typically come from HR system or request form)
NEW_USER = {
"username": "jdoe",
"full_name": "John Doe",
"email": "john.doe@example.com",
"department": "Engineering",
"manager_email": "manager@example.com",
"initial_password": "TemporaryPassword123!" # Should be securely generated & handled
}

# System endpoints and credentials (should be stored securely, not hardcoded!)
AD_API_ENDPOINT = "https://ad-api.example.com/users"
AD_API_KEY = "SECURE_API_KEY_AD"
LINUX_SERVER_LIST = ["linuxdev01.example.com", "linuxprod01.example.com"]
SAAS_APP_API_ENDPOINT = "https://api.saasapp.com/v1/users"
SAAS_APP_API_KEY = "SECURE_API_KEY_SAAS"

# --- Functions for each system ---

def provision_ad_account(user_info):
logging.info(f"Provisioning Active Directory account for {user_info['username']}")
headers = {"Authorization": f"Bearer {AD_API_KEY}", "Content-Type": "application/json"}
payload = {
"userName": user_info['username'],
"displayName": user_info['full_name'],
"mail": user_info['email'],
"department": user_info['department'],
# ... other AD attributes
}
try:
response = requests.post(AD_API_ENDPOINT, headers=headers, json=payload, timeout=30)
response.raise_for_status() # Raise exception for bad status codes (4xx or 5xx)
logging.info(f"AD account created successfully for {user_info['username']}.")
return True
except requests.exceptions.RequestException as e:
logging.error(f"Failed to create AD account for {user_info['username']}: {e}")
return False

def provision_linux_account(user_info, server_list):
logging.info(f"Provisioning Linux accounts for {user_info['username']}")
success_count = 0
for server in server_list:
logging.info(f"Processing Linux server: {server}")
try:
# Use SSH to run useradd command (requires proper SSH setup and permissions)
# This is simplified; real script would handle groups, home dirs, shells etc.
cmd = f"ssh {server} 'sudo useradd -m -c \"{user_info['full_name']}\" {user_info['username']}'"
result = subprocess.run(cmd, shell=True, check=True, capture_output=True, text=True, timeout=60)
logging.info(f"Linux account created on {server} for {user_info['username']}.")
# Need separate step for password setting or SSH key distribution
success_count += 1
except subprocess.CalledProcessError as e:
logging.error(f"Failed to create Linux account on {server}: {e.stderr}")
except subprocess.TimeoutExpired:
logging.error(f"Timeout creating Linux account on {server}")
except Exception as

def process_data(data):
print("Input Data:", data)
normalized_data = normalize(data)
print("Normalized Data:", normalized_data)
return transformed_data

def normalize(data):
max_value = max(data)
if max_value == 0:
return [0] * len(data) # Avoid division by zero
return [x / max_value for x in data]

# db_interface.py

import sqlite3

DATABASE = 'test.db'

def initialize_database():
connection = sqlite3.connect(DATABASE)
cursor = connection.cursor()
cursor.execute('''
CREATE TABLE IF NOT EXISTS users (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL
)
''')
connection.commit()
connection.close()

def add_user(name):
connection = sqlite3.connect(DATABASE)
cursor = connection.cursor()
cursor.execute('INSERT INTO users (name) VALUES (?)', (name,))
connection.commit()
connection.close()

def get_users():
connection = sqlite3.connect(DATABASE)
cursor = connection.cursor()
cursor.execute('SELECT id, name FROM users')
users = cursor.fetchall()
connection.close()
return users

# app.py

from db_interface import initialize_database, add_user, get_users

def setup_app():
# Initialize app dependencies
initialize_database()

def create_user(name):
add_user(name)
return f"User '{name}' successfully created!"

def list_users():
users = get_users()
return users

# test_integration.py

import os
import sqlite3
import unittest
from app import setup_app, create_user, list_users
from db_interface import DATABASE

class IntegrationTest(unittest.TestCase):

@classmethod
def setUpClass(cls):
# Setup application dependencies
# Remove existing test database if it exists to ensure a clean state
if os.path.exists(DATABASE):
os.remove(DATABASE)
setup_app()

def test_create_and_list_user(self):
# Create a new user using the web endpoint
response = create_user("Alice")
self.assertEqual(response, "User 'Alice' successfully created!")

# List users and verify "Alice" is returned
users = list_users()
self.assertTrue(len(users) >= 1)

# Check 'Alice' exists in the users list
user_names = [user[1] for user in users]
self.assertIn("Alice", user_names)

def test_database_persistence(self):
# We insert another user
create_user("Bob")

# Directly access the database to verify persistence
connection = sqlite3.connect(DATABASE)
cursor = connection.cursor()
cursor.execute("SELECT name FROM users WHERE name = ?", ("Bob",))
result = cursor.fetchone()
connection.close()

self.assertIsNotNone(result)
self.assertEqual(result[0], "Bob")

if __name__ == '__main__':
unittest.main()

python -m unittest test_integration.py

# Simple Financial System in Python

class Transaction:
def __init__(self, description, amount):
self.description = description
self.amount = amount

class FinancialSystem:
def __init__(self):
self.transactions = []

def add_income(self, description, amount):
# Adds an income transaction
self.transactions.append(Transaction(description, amount))

def add_expense(self, description, amount):
# Adds an expense transaction (amount is stored as negative)
self.transactions.append(Transaction(description, -abs(amount)))

def generate_profit_and_loss(self):
total_income = sum(t.amount for t in self.transactions if t.amount > 0)
total_expenses = sum(abs(t.amount) for t in self.transactions if t.amount < 0)
net_profit = total_income - total_expenses

return {
'total_income': total_income,
'total_expenses': total_expenses,
'net_profit': net_profit
}

# Example usage
finance = FinancialSystem()
finance.add_income("Salary", 3000)
finance.add_income("Freelance", 1200)
finance.add_expense("Rent", 800)
finance.add_expense("Groceries", 200)

report = finance.generate_profit_and_loss()
print("Total Income:", report['total_income'])
print("Total Expenses:", report['total_expenses'])
print("Net Profit:", report['net_profit'])

# data_processor.py

import pandas as pd
import numpy as np
from typing import List, Dict, Optional
import logging

# Configure logging
logging.basicConfig(level=logging.INFO,
format='%(asctime)s - %(levelname)s: %(message)s')
logger = logging.getLogger(__name__)

class DataProcessor:
"""
A class to process and analyze complex datasets with multiple transformation steps.

Key Responsibilities:
1. Load data from various sources
2. Clean and preprocess the dataset
3. Perform advanced statistical analysis
"""

def __init__(self, data_source: str):
"""
Initialize the DataProcessor with a data source.

Args:
data_source (str): Path to the input data file
"""
self.data_source = data_source
self.raw_data: Optional[pd.DataFrame] = None
self.processed_data: Optional[pd.DataFrame] = None

def load_data(self) -> None:
"""
Load data from the specified source with error handling.
Supports CSV and Excel file formats.
"""
try:
# Determine file type and load accordingly
if self.data_source.endswith('.csv'):
self.raw_data = pd.read_csv(self.data_source)
elif self.data_source.endswith(('.xls', '.xlsx')):
self.raw_data = pd.read_excel(self.data_source)
else:
raise ValueError("Unsupported file format")

logger.info(f"Data loaded successfully. Shape: {self.raw_data.shape}")

except FileNotFoundError:
logger.error(f"File not found: {self.data_source}")
raise
except pd.errors.EmptyDataError:
logger.warning("The data source is empty")
self.raw_data = pd.DataFrame()

def clean_data(self,
drop_columns: Optional[List[str]] = None,
handle_missing: str = 'mean') -> None:
"""
Clean and preprocess the dataset.

Args:
drop_columns (List[str], optional): Columns to drop
handle_missing (str): Strategy for handling missing values
Options: 'mean', 'median', 'drop'
"""
if self.raw_data is None:
logger.error("No data loaded. Call load_data() first.")
return

# Create a copy to preserve original data
self.processed_data = self.raw_data.copy()

# Drop specified columns
if drop_columns:
self.processed_data.drop(columns=drop_columns, inplace=True)

# Handle missing values
if handle_missing == 'mean':
self.processed_data.fillna(self.processed_data.mean(), inplace=True)
elif handle_missing == 'median':
self.processed_data.fillna(self.processed_data.median(), inplace=True)
elif handle_missing == 'drop':
self.processed_data.dropna(inplace=True)

logger.info(f"Data cleaned. Remaining shape: {self.processed_data.shape}")

def analyze_data(self) -> Dict[str, float]:
"""
Perform statistical analysis on the processed data.

Returns:
Dict[str, float]: Key statistical metrics
"""
if self.processed_data is None:
logger.error("No processed data available")
return {}

# Compute advanced statistical metrics
analysis_results = {
'mean': self.processed_data.mean().to_dict(),
'median': self.processed_data.median().to_dict(),
'std_dev': self.processed_data.std().to_dict(),
'correlation_matrix': self.processed_data.corr().to_dict()
}

return analysis_results

def detect_outliers(self, method: str = 'iqr') -> pd.DataFrame:
"""
Detect outliers using different statistical methods.

Args:
method (str): Outlier detection method

Returns:
pd.DataFrame: Detected outliers
"""
if self.processed_data is None:
logger.error("No processed data available")
return pd.DataFrame()

if method == 'iqr':
# Interquartile Range (IQR) method
Q1 = self.processed_data.quantile(0.25)
Q3 = self.processed_data.quantile(0.75)
IQR = Q3 - Q1

outliers = ((self.processed_data < (Q1 - 1.5 * IQR)) |
(self.processed_data > (Q3 + 1.5 * IQR)))

return self.processed_data[outliers.any(axis=1)]

logger.warning(f"Unsupported outlier detection method: {method}")
return pd.DataFrame()

def main():
"""
Main execution function demonstrating the DataProcessor workflow.
"""
try:
# Initialize processor
processor = DataProcessor('sample_data.csv')

# Load and process data
processor.load_data()
processor.clean_data(drop_columns=['unnecessary_column'])

# Perform analysis
analysis_results = processor.analyze_data()
print("Analysis Results:", analysis_results)

# Detect outliers
outliers = processor.detect_outliers()
print("Outliers:", outliers)

except Exception as e:
logger.error(f"An error occurred: {e}")

if __name__ == "__main__":
main()

import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import seaborn as sns
from scipy import stats
from typing import Dict, List, Tuple
import logging

# Configure logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

class SalesAnalyzer:
"""
Analyzes sales data to provide insights on trends, patterns, and statistics.
"""

def __init__(self, data_path: str):
"""
Initialize the analyzer with data path.

Args:
data_path (str): Path to the sales data file
"""
self.data_path = data_path
self.df = None
self.summary_stats = {}

def load_and_prepare_data(self) -> None:
"""
Load data and perform initial preprocessing steps.
"""
try:
# Load the data
self.df = pd.read_csv(self.data_path)

# Convert date column to datetime
self.df['date'] = pd.to_datetime(self.df['date'])

# Basic data cleaning
self.df = self.df.dropna()

logger.info(f"Data loaded successfully. Shape: {self.df.shape}")

except Exception as e:
logger.error(f"Error loading data: {str(e)}")
raise

def calculate_basic_stats(self) -> Dict:
"""
Calculate basic statistical measures for numerical columns.

Returns:
Dict: Dictionary containing basic statistics
"""
numeric_cols = self.df.select_dtypes(include=[np.number]).columns

stats_dict = {}
for col in numeric_cols:
stats_dict[col] = {
'mean': self.df[col].mean(),
'median': self.df[col].median(),
'std': self.df[col].std(),
'min': self.df[col].min(),
'max': self.df[col].max()
}

self.summary_stats = stats_dict
return stats_dict

def analyze_sales_trends(self) -> pd.DataFrame:
"""
Analyze monthly sales trends.

Returns:
pd.DataFrame: Monthly sales summary
"""
# Group by month and calculate statistics
monthly_sales = self.df.groupby(self.df['date'].dt.to_period('M')).agg({
'sales': ['sum', 'mean', 'count'],
'revenue': ['sum', 'mean']
}).round(2)

# Calculate month-over-month growth
monthly_sales['sales_growth'] = (
monthly_sales[('sales', 'sum')].pct_change() * 100
).round(2)

return monthly_sales

def perform_customer_segmentation(self) -> Dict:
"""
Segment customers based on purchase behavior.

Returns:
Dict: Customer segments and their characteristics
"""
customer_stats = self.df.groupby('customer_id').agg({
'sales': ['count', 'sum', 'mean'],
'revenue': 'sum'
})

# Define segments using quantiles
segments = {
'High Value': customer_stats[('revenue', 'sum')] >= customer_stats[('revenue', 'sum')].quantile(0.75),
'Medium Value': (customer_stats[('revenue', 'sum')] >= customer_stats[('revenue', 'sum')].quantile(0.25)) &
(customer_stats[('revenue', 'sum')] < customer_stats[('revenue', 'sum')].quantile(0.75)),
'Low Value': customer_stats[('revenue', 'sum')] < customer_stats[('revenue', 'sum')].quantile(0.25)
}

segment_stats = {}
for segment_name, segment_mask in segments.items():
segment_stats[segment_name] = {
'count': segment_mask.sum(),
'avg_revenue': customer_stats.loc[segment_mask, ('revenue', 'sum')].mean(),
'avg_purchases': customer_stats.loc[segment_mask, ('sales', 'count')].mean()
}

return segment_stats

def visualize_trends(self, save_path: str = None) -> None:
"""
Create visualizations for key metrics.

Args:
save_path (str, optional): Path to save the visualizations
"""
# Set up the plotting style
plt.style.use('seaborn')

# Create a figure with multiple subplots
fig, axes = plt.subplots(2, 2, figsize=(15, 12))

# 1. Monthly Sales Trend
monthly_sales = self.analyze_sales_trends()
monthly_sales[('sales', 'sum')].plot(
ax=axes[0, 0],
title='Monthly Sales Trend'
)
axes[0, 0].set_xlabel('Month')
axes[0, 0].set_ylabel('Total Sales')

# 2. Sales Distribution
sns.histplot(
data=self.df,
x='sales',
ax=axes[0, 1],
bins=30
)
axes[0, 1].set_title('Sales Distribution')

# 3. Revenue by Customer Segment
customer_segments = self.perform_customer_segmentation()
segment_data = pd.DataFrame(customer_segments).T
segment_data['avg_revenue'].plot(
kind='bar',
ax=axes[1, 0],
title='Average Revenue by Customer Segment'
)
axes[1, 0].set_ylabel('Average Revenue')

# 4. Correlation Heatmap
numeric_cols = self.df.select_dtypes(include=[np.number]).columns
sns.heatmap(
self.df[numeric_cols].corr(),
annot=True,
cmap='coolwarm',
ax=axes[1, 1]
)
axes[1, 1].set_title('Correlation Heatmap')

plt.tight_layout()

if save_path:
plt.savefig(save_path)
logger.info(f"Visualizations saved to {save_path}")

plt.show()

def generate_report(self) -> str:
"""
Generate a summary report of the analysis.

Returns:
str: Formatted report string
"""
report = []
report.append("=== Sales Analysis Report ===\n")

# Basic Stats
report.append("Basic Statistics:")
for metric, stats in self.summary_stats.items():
report.append(f"\n{metric}:")
for stat_name, value in stats.items():
report.append(f" {stat_name}: {value:.2f}")

# Sales Trends
monthly_sales = self.analyze_sales_trends()
report.append("\n\nSales Trends:")
report.append(f"Total Sales: {monthly_sales[('sales', 'sum')].sum():,.2f}")
report.append(f"Average Monthly Sales: {monthly_sales[('sales', 'mean')].mean():,.2f}")

# Customer Segments
segments = self.perform_customer_segmentation()
report.append("\n\nCustomer Segments:")
for segment, stats in segments.items():
report.append(f"\n{segment}:")
for stat_name, value in stats.items():
report.append(f" {stat_name}: {value:,.2f}")

return "\n".join(report)

def main():
"""
Main execution function.
"""
try:
# Initialize analyzer
analyzer = SalesAnalyzer('sales_data.csv')

# Perform analysis
analyzer.load_and_prepare_data()
analyzer.calculate_basic_stats()

# Generate visualizations
analyzer.visualize_trends('sales_analysis_plots.png')

# Generate and print report
report = analyzer.generate_report()
print(report)

# Save report to file
with open('sales_analysis_report.txt', 'w') as f:
f.write(report)

logger.info("Analysis completed successfully")

except Exception as e:
logger.error(f"Error during analysis: {str(e)}")
raise

if __name__ == "__main__":
main()

import logging
import time
import tracemalloc
import psutil
import requests
from datetime import datetime
from functools import wraps

# Configure logging
logging.basicConfig(
level=logging.DEBUG,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
filename='app_debug.log',
filemode='w'
)
logger = logging.getLogger('performance_debugger')

# Performance monitoring decorator
def performance_monitor(func):
@wraps(func)
def wrapper(*args, **kwargs):
# Start memory tracking
tracemalloc.start()
process = psutil.Process()
memory_before = process.memory_info().rss / 1024 / 1024 # MB

# Start timing
start_time = time.time()
logger.info(f"Starting {func.__name__} at {datetime.now()}")

# Execute function
try:
result = func(*args, **kwargs)

# Log success
logger.info(f"Successfully executed {func.__name__}")
return result
except Exception as e:
# Log error with traceback
logger.error(f"Error in {func.__name__}: {str(e)}")
logger.error(traceback.format_exc())
raise
finally:
# Measure execution time
execution_time = time.time() - start_time

# Measure memory usage
memory_after = process.memory_info().rss / 1024 / 1024 # MB
memory_diff = memory_after - memory_before

# Get memory snapshot
current, peak = tracemalloc.get_traced_memory()
tracemalloc.stop()

# Log performance metrics
logger.info(f"Function: {func.__name__}")
logger.info(f"Execution time: {execution_time:.4f} seconds")
logger.info(f"Memory change: {memory_diff:.2f} MB")
logger.info(f"Current memory usage: {current / 1024 / 1024:.2f} MB")
logger.info(f"Peak memory usage: {peak / 1024 / 1024:.2f} MB")

# Log top memory consumers if significant memory increase
if memory_diff > 10: # If more than 10MB increase
logger.warning("Significant memory increase detected!")
snapshot = tracemalloc.take_snapshot()
top_stats = snapshot.statistics('lineno')
logger.warning("Top 10 memory consumers:")
for stat in top_stats[:10]:
logger.warning(f"{stat}")

return wrapper

# Database connection monitoring
class DatabaseMonitor:
def __init__(self, connection_string):
self.connection_string = connection_string
self.query_times = []
self.slow_threshold = 0.5 # seconds

def execute_query(self, query, params=None):
start_time = time.time()
logger.debug(f"Executing query: {query}")
logger.debug(f"With parameters: {params}")

try:
# Simulate database connection and query execution
# In real code, this would use actual database driver
time.sleep(0.1) # Simulating query execution time
result = {"data": "sample data"} # Simulated result

execution_time = time.time() - start_time
self.query_times.append(execution_time)

if execution_time > self.slow_threshold:
logger.warning(f"Slow query detected! Time: {execution_time:.4f}s")
logger.warning(f"Query: {query}")
logger.warning(f"Parameters: {params}")

# Suggest query optimization
self.suggest_optimization(query)

return result
except Exception as e:
logger.error(f"Database error: {str(e)}")
logger.error(traceback.format_exc())
raise

def suggest_optimization(self, query):
# Simple query optimization suggestions
if "SELECT *" in query:
logger.warning("Optimization suggestion: Avoid using SELECT * - specify only needed columns")

if "WHERE" not in query and ("SELECT" in query and "FROM" in query):
logger.warning("Optimization suggestion: Query has no WHERE clause - might return too many rows")

if "JOIN" in query and "INDEX" not in query:
logger.warning("Optimization suggestion: Consider adding indexes for JOIN operations")

# API request debugging
class APIDebugger:
def __init__(self):
self.request_history = []

def make_request(self, url, method="GET", headers=None, data=None, timeout=10):
request_id = len(self.request_history) + 1
logger.info(f"API Request #{request_id} - {method} {url}")

if headers:
logger.debug(f"Request #{request_id} Headers: {headers}")
if data:
logger.debug(f"Request #{request_id} Data: {data}")

start_time = time.time()

try:
# In real code, this would make an actual HTTP request
# response = requests.request(method, url, headers=headers, json=data, timeout=timeout)

# Simulate API request
time.sleep(0.2)
response_status = 200
response_data = {"status": "success", "data": {"id": 123}}

# Record request details
request_info = {
"id": request_id,
"url": url,
"method": method,
"headers": headers,
"data": data,
"status_code": response_status,
"response_time": time.time() - start_time,
"timestamp": datetime.now()
}
self.request_history.append(request_info)

# Log response
logger.info(f"API Request #{request_id} completed - Status: {response_status}")
logger.debug(f"Response data: {response_data}")

if request_info["response_time"] > 1.0:
logger.warning(f"Slow API request detected! Time: {request_info['response_time']:.4f}s")

return response_data

except Exception as e:
logger.error(f"API Request #{request_id} failed: {str(e)}")
logger.error(traceback.format_exc())
raise

# Frontend performance monitoring
class FrontendPerformanceMonitor:
def __init__(self):
self.render_times = {}

def log_render_time(self, component_name, render_time):
if component_name not in self.render_times:
self.render_times[component_name] = []

self.render_times[component_name].append(render_time)

if render_time > 0.1: # 100ms threshold
logger.warning(f"Slow component render: {component_name} took {render_time*1000:.2f}ms")

def analyze_performance(self):
logger.info("Frontend Performance Analysis:")

for component, times in self.render_times.items():
avg_time = sum(times) / len(times)
max_time = max(times)

logger.info(f"Component: {component}")
logger.info(f" Average render time: {avg_time*1000:.2f}ms")
logger.info(f" Maximum render time: {max_time*1000:.2f}ms")
logger.info(f" Render count: {len(times)}")

if avg_time > 0.05: # 50ms threshold
logger.warning(f"Component {component} has high average render time")
logger.warning("Suggestion: Consider memoization or component optimization")

# Example usage of the debugging tools
@performance_monitor
def process_user_data(user_id):
logger.info(f"Processing data for user {user_id}")

# Simulate database operations
db = DatabaseMonitor("postgresql://user:password@localhost:5432/mydb")
user_data = db.execute_query("SELECT * FROM users WHERE user_id = %s", (user_id,))

# Simulate API calls
api = APIDebugger()
external_data = api.make_request(f"https://api.example.com/users/{user_id}")

# Simulate frontend rendering
frontend = FrontendPerformanceMonitor()
frontend.log_render_time("UserProfile", 0.08)
frontend.log_render_time("UserActivityFeed", 0.15)
frontend.log_render_time("UserSettings", 0.03)

# Analyze frontend performance
frontend.analyze_performance()

# Simulate some memory-intensive operation
large_list = [i for i in range(1000000)]

# Return processed data
return {
"user_data": user_data,
"external_data": external_data,
"processed_items": len(large_list)
}

# Run the function with debugging
if __name__ == "__main__":
try:
logger.info("Starting application debugging session")
result = process_user_data(12345)
logger.info(f"Process completed successfully: {result}")
except Exception as e:
logger.critical(f"Application failed: {str(e)}")
finally:
logger.info("Debugging session ended")

# app.py
from flask import Flask, render_template, redirect, url_for, request, flash, jsonify
from flask_sqlalchemy import SQLAlchemy
from flask_login import LoginManager, UserMixin, login_user, login_required, logout_user, current_user
from flask_wtf import FlaskForm
from wtforms import StringField, PasswordField, SubmitField
from wtforms.validators import DataRequired, Email, Length
from werkzeug.security import generate_password_hash, check_password_hash
import os

# Initialize Flask application
app = Flask(__name__)
app.config['SECRET_KEY'] = os.environ.get('SECRET_KEY', 'dev-key-123')
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///site.db'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False

# Initialize extensions
db = SQLAlchemy(app)
login_manager = LoginManager(app)
login_manager.login_view = 'login'

# Database Models
class User(db.Model, UserMixin):
id = db.Column(db.Integer, primary_key=True)
username = db.Column(db.String(20), unique=True, nullable=False)
email = db.Column(db.String(120), unique=True, nullable=False)
password_hash = db.Column(db.String(128))
posts = db.relationship('Post', backref='author', lazy=True)

def set_password(self, password):
self.password_hash = generate_password_hash(password)

def check_password(self, password):
return check_password_hash(self.password_hash, password)

class Post(db.Model):
id = db.Column(db.Integer, primary_key=True)
title = db.Column(db.String(100), nullable=False)
content = db.Column(db.Text, nullable=False)
user_id = db.Column(db.Integer, db.ForeignKey('user.id'), nullable=False)

# Forms
class RegistrationForm(FlaskForm):
username = StringField('Username', validators=[DataRequired(), Length(min=4, max=20)])
email = StringField('Email', validators=[DataRequired(), Email()])
password = PasswordField('Password', validators=[DataRequired(), Length(min=6)])
submit = SubmitField('Sign Up')

class LoginForm(FlaskForm):
email = StringField('Email', validators=[DataRequired(), Email()])
password = PasswordField('Password', validators=[DataRequired()])
submit = SubmitField('Login')

class PostForm(FlaskForm):
title = StringField('Title', validators=[DataRequired()])
content = StringField('Content', validators=[DataRequired()])
submit = SubmitField('Create Post')

# Authentication Routes
@app.route('/register', methods=['GET', 'POST'])
def register():
if current_user.is_authenticated:
return redirect(url_for('dashboard'))

form = RegistrationForm()
if form.validate_on_submit():
user = User(username=form.username.data, email=form.email.data)
user.set_password(form.password.data)
db.session.add(user)
db.session.commit()
flash('Account created successfully!', 'success')
return redirect(url_for('login'))

return render_template('register.html', form=form)

@app.route('/login', methods=['GET', 'POST'])
def login():
if current_user.is_authenticated:
return redirect(url_for('dashboard'))

form = LoginForm()
if form.validate_on_submit():
user = User.query.filter_by(email=form.email.data).first()
if user and user.check_password(form.password.data):
login_user(user)
return redirect(url_for('dashboard'))
flash('Invalid email or password', 'danger')
return render_template('login.html', form=form)

@app.route('/logout')
@login_required
def logout():
logout_user()
return redirect(url_for('home'))

# Main Application Routes
@app.route('/')
def home():
posts = Post.query.order_by(Post.id.desc()).limit(3).all()
return render_template('home.html', posts=posts)

@app.route('/dashboard')
@login_required
def dashboard():
posts = current_user.posts
return render_template('dashboard.html', posts=posts)

@app.route('/post/new', methods=['GET', 'POST'])
@login_required
def new_post():
form = PostForm()
if form.validate_on_submit():
post = Post(title=form.title.data, content=form.content.data, author=current_user)
db.session.add(post)
db.session.commit()
flash('Post created successfully!', 'success')
return redirect(url_for('dashboard'))
return render_template('create_post.html', form=form)

# API Endpoints
@app.route('/api/posts', methods=['GET'])
def get_posts():
posts = Post.query.all()
return jsonify([{
'id': post.id,
'title': post.title,
'content': post.content,
'author': post.author.username
} for post in posts])

@app.route('/api/posts/<int:post_id>', methods=['GET'])
def get_post(post_id):
post = Post.query.get_or_404(post_id)
return jsonify({
'id': post.id,
'title': post.title,
'content': post.content,
'author': post.author.username
})

# Error Handlers
@app.errorhandler(404)
def page_not_found(e):
return render_template('404.html'), 404

@app.errorhandler(500)
def internal_error(e):
db.session.rollback()
return render_template('500.html'), 500

# Login Manager Loader
@login_manager.user_loader
def load_user(user_id):
return User.query.get(int(user_id))

# Create Database Tables
with app.app_context():
db.create_all()

if __name__ == '__main__':
app.run(debug=True)

templates/
├── base.html
├── home.html
├── register.html
├── login.html
├── dashboard.html
├── create_post.html
├── 404.html
└── 500.html

<!DOCTYPE html>
<html>
<head>
<meta charset="utf-8">
<title>{% block title %}Flask App{% endblock %}</title>
<link rel="stylesheet" href="https://stackpath.bootstrapcdn.com/bootstrap/4.3.1/css/bootstrap.min.css">
</head>
<body>
<nav class="navbar navbar-expand-lg navbar-dark bg-dark">
<a class="navbar-brand" href="{{ url_for('home') }}">FlaskApp</a>
<div class="collapse navbar-collapse">
<div class="navbar-nav">
{% if current_user.is_authenticated %}
<a class="nav-item nav-link" href="{{ url_for('dashboard') }}">Dashboard</a>
<a class="nav-item nav-link" href="{{ url_for('logout') }}">Logout</a>
{% else %}
<a class="nav-item nav-link" href="{{ url_for('login') }}">Login</a>
<a class="nav-item nav-link" href="{{ url_for('register') }}">Register</a>
{% endif %}
</div>
</div>
</nav>
<div class="container mt-4">
{% with messages = get_flashed_messages(with_categories=true) %}
{% if messages %}
{% for category, message in messages %}
<div class="alert alert-{{ category }}">{{ message }}</div>
{% endfor %}
{% endif %}
{% endwith %}
{% block content %}{% endblock %}
</div>
</body>
</html>

pip install flask flask-sqlalchemy flask-login flask-wtf python-dotemail

python3 -c "from app import app, db; with app.app_context(): db.create_all()"

FLASK_APP=app.py flask run

# app.py
from flask import Flask, render_template
from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///site.db'
db = SQLAlchemy(app)

# Models
class User(db.Model):
id = db.Column(db.Integer, primary_key=True)
username = db.Column(db.String(20), unique=True, nullable=False)

# Routes
@app.route('/')
def home():
return render_template('home.html')

if __name__ == '__main__':
app.run(debug=True)

pip install Flask firebase-admin

from flask import Flask, request, jsonify
import firebase_admin
from firebase_admin import credentials, firestore

# Initialize Flask app
app = Flask(__name__)

# Initialize Firebase Admin SDK
cred = credentials.Certificate('path/to/your/serviceAccountKey.json')
firebase_admin.initialize_app(cred)
db = firestore.client()

# Define a collection name
COLLECTION_NAME = 'items'

@app.route('/items', methods=['GET'])
def get_items():
try:
items_ref = db.collection(COLLECTION_NAME)
docs = items_ref.stream()
items = [doc.to_dict() for doc in docs]
return jsonify(items), 200
except Exception as e:
return jsonify({'error': str(e)}), 500

@app.route('/items', methods=['POST'])
def create_item():
try:
data = request.get_json()
if not data:
return jsonify({'error': 'No data provided'}), 400
db.collection(COLLECTION_NAME).add(data)
return jsonify({'message': 'Item created successfully'}), 201
except Exception as e:
return jsonify({'error': str(e)}), 500

@app.route('/items/<item_id>', methods=['GET'])
def get_item(item_id):
try:
item_ref = db.collection(COLLECTION_NAME).document(item_id)
item = item_ref.get()
if item.exists:
return jsonify(item.to_dict()), 200
else:
return jsonify({'error': 'Item not found'}), 404
except Exception as e:
return jsonify({'error': str(e)}), 500

@app.route('/items/<item_id>', methods=['PUT'])
def update_item(item_id):
try:
data = request.get_json()
if not data:
return jsonify({'error': 'No data provided'}), 400
item_ref = db.collection(COLLECTION_NAME).document(item_id)
item_ref.update(data)
return jsonify({'message': 'Item updated successfully'}), 200
except Exception as e:
return jsonify({'error': str(e)}), 500

@app.route('/items/<item_id>', methods=['DELETE'])
def delete_item(item_id):
try:
item_ref = db.collection(COLLECTION_NAME).document(item_id)
item_ref.delete()
return jsonify({'message': 'Item deleted successfully'}), 200
except Exception as e:
return jsonify({'error': str(e)}), 500

if __name__ == '__main__':
app.run(debug=True)

python app.py