Add user ""

src/__init__.py

@@ -0,0 +1,5 @@
+__version__ = "1.0.0"
+
+from .utils import setup_logger, get_config
+
+__all__ = ["setup_logger", "get_config"]

src/main.py

@@ -0,0 +1,20 @@
+from flask import Flask
+from src.utils import setup_logger, get_config
+
+# Initialize app and logger
+app = Flask(__name__)
+logger = setup_logger("main")
+
+@app.route("/")
+def home():
+    """Home endpoint."""
+    logger.info("Home endpoint was accessed.")
+    return {"message": "Welcome to the API!"}
+
+if __name__ == "__main__":
+    # Load configuration
+    port = int(get_config("APP_PORT", 5000))
+    debug = get_config("DEBUG", "false").lower() == "true"
+
+    logger.info(f"Starting application on port {port}")
+    app.run(host="0.0.0.0", port=port, debug=debug)

src/services/__init__.py

@@ -0,0 +1,14 @@
+from .payment_service import process_payment, refund_payment
+from .email_service import send_email, send_bulk_emails
+from .external_api_service import fetch_external_data
+from .background_jobs import schedule_task, run_scheduled_jobs
+
+__all__ = [
+    "process_payment",
+    "refund_payment",
+    "send_email",
+    "send_bulk_emails",
+    "fetch_external_data",
+    "schedule_task",
+    "run_scheduled_jobs",
+]

src/services/background_jobs.py

@@ -0,0 +1,31 @@
+from celery import Celery
+
+# Configure Celery
+app = Celery("tasks", broker="redis://localhost:6379/0")
+
+@app.task
+def send_email_task(to_email, subject, body):
+    """
+    Background task to send an email.
+    """
+    from .email_service import send_email
+    send_email(to_email, subject, body)
+    print(f"Background email sent to {to_email}")
+
+@app.task
+def schedule_task(task_name, delay):
+    """
+    Simulates scheduling a task with a delay.
+    """
+    import time
+    print(f"Scheduling task '{task_name}' to run after {delay} seconds...")
+    time.sleep(delay)
+    print(f"Task '{task_name}' executed.")
+
+def run_scheduled_jobs():
+    """
+    Runs all scheduled background jobs.
+    """
+    # Example usage
+    schedule_task.apply_async(("test_task", 5))
+    send_email_task.apply_async(("[email protected]", "Hello", "This is a test email."))

src/services/email_service.py

@@ -0,0 +1,36 @@
+import smtplib
+from email.mime.text import MIMEText
+from email.mime.multipart import MIMEMultipart
+
+SMTP_SERVER = "smtp.gmail.com"
+SMTP_PORT = 587
+SMTP_USERNAME = "[email protected]"
+SMTP_PASSWORD = "your-email-password"
+
+def send_email(to_email, subject, body):
+    """
+    Sends a single email.
+    """
+    try:
+        msg = MIMEMultipart()
+        msg["From"] = SMTP_USERNAME
+        msg["To"] = to_email
+        msg["Subject"] = subject
+
+        msg.attach(MIMEText(body, "plain"))
+
+        with smtplib.SMTP(SMTP_SERVER, SMTP_PORT) as server:
+            server.starttls()
+            server.login(SMTP_USERNAME, SMTP_PASSWORD)
+            server.send_message(msg)
+
+        print(f"Email sent to {to_email}")
+    except Exception as e:
+        print(f"Error sending email: {e}")
+
+def send_bulk_emails(email_list, subject, body):
+    """
+    Sends bulk emails to a list of recipients.
+    """
+    for email in email_list:
+        send_email(email, subject, body)

src/services/external_api_service.py

@@ -0,0 +1,25 @@
+import requests
+
+def fetch_external_data(api_url, params=None, headers=None):
+    """
+    Fetches data from an external API.
+    """
+    try:
+        response = requests.get(api_url, params=params, headers=headers)
+        response.raise_for_status()  # Raises an HTTPError for bad responses
+        return response.json()
+    except requests.exceptions.RequestException as e:
+        print(f"Error fetching data from API: {e}")
+        return None
+
+def post_to_external_api(api_url, data=None, headers=None):
+    """
+    Sends data to an external API.
+    """
+    try:
+        response = requests.post(api_url, json=data, headers=headers)
+        response.raise_for_status()
+        return response.json()
+    except requests.exceptions.RequestException as e:
+        print(f"Error posting data to API: {e}")
+        return None

src/services/payment_service.py

@@ -0,0 +1,31 @@
+import stripe
+
+# Stripe API configuration
+stripe.api_key = "your-secret-key"
+
+def process_payment(amount, currency="usd", source=None, description=""):
+    """
+    Processes a payment using Stripe.
+    """
+    try:
+        charge = stripe.Charge.create(
+            amount=int(amount * 100),  # Stripe uses the smallest currency unit
+            currency=currency,
+            source=source,
+            description=description,
+        )
+        return charge
+    except stripe.error.StripeError as e:
+        print(f"Error processing payment: {e}")
+        return None
+
+def refund_payment(charge_id):
+    """
+    Refunds a payment using Stripe.
+    """
+    try:
+        refund = stripe.Refund.create(charge=charge_id)
+        return refund
+    except stripe.error.StripeError as e:
+        print(f"Error refunding payment: {e}")
+        return None

src/utils.py

@@ -0,0 +1,21 @@
+import logging
+import os
+
+def setup_logger(name: str, level: int = logging.INFO) -> logging.Logger:
+    """Set up a logger with the given name and logging level."""
+    logger = logging.getLogger(name)
+    logger.setLevel(level)
+
+    # Create console handler
+    if not logger.handlers:  # Avoid duplicate handlers
+        ch = logging.StreamHandler()
+        ch.setLevel(level)
+        formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
+        ch.setFormatter(formatter)
+        logger.addHandler(ch)
+
+    return logger
+
+def get_config(key: str, default: str = None) -> str:
+    """Get a configuration value from environment variables."""
+    return os.getenv(key, default)

src/utils/__init__.py

@@ -0,0 +1,16 @@
+from .validation import validate_email, validate_password
+from .auth_utils import generate_token, verify_token
+from .logger import setup_logger
+from .date_utils import format_date, get_current_timestamp
+from .config import get_config
+
+__all__ = [
+    "validate_email",
+    "validate_password",
+    "generate_token",
+    "verify_token",
+    "setup_logger",
+    "format_date",
+    "get_current_timestamp",
+    "get_config",
+]

src/utils/auth_utils.py

@@ -0,0 +1,24 @@
+import jwt
+import datetime
+from typing import Union
+
+SECRET_KEY = "your-secret-key"
+
+def generate_token(user_id: int, expires_in: int = 3600) -> str:
+    """Generate a JWT token for a user."""
+    payload = {
+        "user_id": user_id,
+        "exp": datetime.datetime.utcnow() + datetime.timedelta(seconds=expires_in),
+    }
+    return jwt.encode(payload, SECRET_KEY, algorithm="HS256")
+
+def verify_token(token: str) -> Union[dict, None]:
+    """Verify a JWT token and return its payload."""
+    try:
+        payload = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
+        return payload
+    except jwt.ExpiredSignatureError:
+        print("Token has expired.")
+    except jwt.InvalidTokenError:
+        print("Invalid token.")
+    return None

src/utils/config.py

@@ -0,0 +1,5 @@
+import os
+
+def get_config(key: str, default: str = None) -> str:
+    """Get a configuration value from environment variables."""
+    return os.getenv(key, default)

src/utils/date_utils.py

@@ -0,0 +1,9 @@
+from datetime import datetime, timezone
+
+def get_current_timestamp() -> int:
+    """Return the current timestamp in seconds."""
+    return int(datetime.now(tz=timezone.utc).timestamp())
+
+def format_date(date: datetime, format_string: str = "%Y-%m-%d %H:%M:%S") -> str:
+    """Format a datetime object as a string."""
+    return date.strftime(format_string)

src/utils/logger.py

@@ -0,0 +1,20 @@
+import logging
+
+def setup_logger(name: str, level: int = logging.INFO) -> logging.Logger:
+    """Set up a logger with the given name and logging level."""
+    logger = logging.getLogger(name)
+    logger.setLevel(level)
+
+    # Create console handler
+    ch = logging.StreamHandler()
+    ch.setLevel(level)
+
+    # Create formatter and add it to the handler
+    formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
+    ch.setFormatter(formatter)
+
+    # Add the handler to the logger
+    if not logger.handlers:
+        logger.addHandler(ch)
+
+    return logger

src/utils/validation.py

@@ -0,0 +1,16 @@
+import re
+
+def validate_email(email: str) -> bool:
+    """Validate if the provided string is a valid email address."""
+    email_regex = r'^[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+$'
+    return re.match(email_regex, email) is not None
+
+def validate_password(password: str) -> bool:
+    """Validate password strength (e.g., at least 8 characters, contains a number)."""
+    if len(password) < 8:
+        return False
+    if not any(char.isdigit() for char in password):
+        return False
+    if not any(char.isalpha() for char in password):
+        return False
+    return True