SQLite Relational Database Tutorials

"""
SQLite Connection Demonstration Module.

This script demonstrates connecting to an SQLite database, initializing a cursor,
and safely closing the database connection.

Note: Do NOT name python files 'sqlite3.py' to prevent shadowing the standard library module.
"""

import sqlite3
from typing import Optional


def connect_database(db_name: str = "mydatabase.db") -> Optional[sqlite3.Connection]:
    """Connect to SQLite database file.

    Args:
        db_name (str): Database file path or name.

    Returns:
        Optional[sqlite3.Connection]: SQLite connection object if successful.
    """
    try:
        # Establish connection to the database
        connection = sqlite3.connect(db_name)
        # Initialize cursor for executing queries
        cursor = connection.cursor()
        print(f"Successfully connected to database '{db_name}'.")

        # Close database connection
        connection.close()
        print("Database connection closed successfully.")
        return connection
    except sqlite3.Error as error:
        print(f"Error connecting to database: {error}")
        return None


def create_connection(db_name: str = "mydatabase.db"):
    """Create and return an active SQLite connection and cursor.

    Args:
        db_name (str): Database file path or name.

    Returns:
        Tuple[sqlite3.Connection, sqlite3.Cursor]: Active connection and cursor tuple.
    """
    connection = sqlite3.connect(db_name)
    cursor = connection.cursor()
    return connection, cursor


def close_connection(connection: sqlite3.Connection) -> None:
    """Safely close active SQLite connection."""
    if connection:
        connection.close()


if __name__ == "__main__":
    connect_database()

import json
from typing import Dict, Any, Tuple

class RESTStudioRunner:
    """In-memory REST API service controller."""
    def __init__(self):
        self.services = {
            1: {"id": 1, "name": "PostgreSQL Database Service", "category": "Database", "status": "active"},
            2: {"id": 2, "name": "Redis Cache Cluster", "category": "Caching", "status": "active"},
            3: {"id": 3, "name": "RabbitMQ Message Queue", "category": "Messaging", "status": "idle"}
        }

    def handle_request(self, method: str, path: str, payload: Dict[str, Any] = None) -> Tuple[int, Dict[str, Any]]:
        """Route HTTP method and path to handler."""
        if method == "GET" and path == "/api/v1/services":
            return 200, {"status_code": 200, "message": "OK", "count": len(self.services), "data": list(self.services.values())}

        elif method == "GET" and path.startswith("/api/v1/services/"):
            try:
                service_id = int(path.split("/")[-1])
                service = self.services.get(service_id)
                if service:
                    return 200, {"status_code": 200, "data": service}
                return 404, {"status_code": 404, "error": f"Service #{service_id} Not Found"}
            except ValueError:
                return 400, {"status_code": 400, "error": "Invalid Service ID"}

        elif method == "POST" and path == "/api/v1/services":
            if not payload or "name" not in payload:
                return 400, {"status_code": 400, "error": "Field 'name' is required"}
            new_id = max(self.services.keys(), default=0) + 1
            new_service = {
                "id": new_id,
                "name": payload["name"],
                "category": payload.get("category", "General"),
                "status": "active",
            }
            self.services[new_id] = new_service
            return 201, {
                "status_code": 201,
                "message": "Resource Created Successfully",
                "data": new_service,
            }

        elif method == "PUT" and path.startswith("/api/v1/services/"):
            service_id = int(path.split("/")[-1])
            if service_id not in self.services:
                return 404, {"status_code": 404, "error": f"Service #{service_id} Not Found"}
            if payload:
                self.services[service_id].update(payload)
            return 200, {
                "status_code": 200,
                "message": "Resource Updated Successfully",
                "data": self.services[service_id],
            }

        elif method == "DELETE" and path.startswith("/api/v1/services/"):
            service_id = int(path.split("/")[-1])
            if service_id in self.services:
                deleted = self.services.pop(service_id)
                return 200, {
                    "status_code": 200,
                    "message": "Resource Deleted Successfully",
                    "deleted_id": service_id,
                    "deleted_name": deleted["name"],
                }
            return 404, {"status_code": 404, "error": f"Service #{service_id} Not Found"}

        return 400, {"status_code": 400, "error": f"Unsupported HTTP Method '{method}'"}