Python Full Stack Web Developer Course

Real-World Examples

Let's explore some real-world examples of encapsulation in Python web development contexts. Create a file named real_world_examples.py with the following code:

# File: encapsulation/real_world_examples.py

# Example 1: Database Connection Pool
class DatabaseConnectionPool:
    """A simplified database connection pool demonstrating encapsulation"""
    
    def __init__(self, host, port, user, password, database, max_connections=5):
        # Store connection parameters as protected attributes
        self._host = host
        self._port = port
        self._user = user
        self._password = password  # In a real scenario, this would be more securely handled
        self._database = database
        self._max_connections = max_connections
        
        # Private attributes for internal state
        self.__connections = []
        self.__active_connections = 0
    
    def get_connection(self):
        """Get a connection from the pool"""
        if self.__active_connections >= self._max_connections:
            raise Exception("Connection pool exhausted")
        
        # Create a new connection if needed
        if len(self.__connections) == 0:
            connection = self.__create_connection()
        else:
            connection = self.__connections.pop()
        
        self.__active_connections += 1
        return connection
    
    def release_connection(self, connection):
        """Return a connection to the pool"""
        if connection is None:
            return
        
        # In a real scenario, we would validate that the connection belongs to this pool
        self.__connections.append(connection)
        self.__active_connections -= 1
    
    def __create_connection(self):
        """Private method to create a new database connection"""
        print(f"Creating new connection to {self._host}:{self._port}/{self._database}")
        # In a real scenario, this would use a database driver to create a connection
        return {
            "id": len(self.__connections) + self.__active_connections + 1,
            "host": self._host,
            "database": self._database
        }
    
    @property
    def available_connections(self):
        """Get the number of available connections"""
        return len(self.__connections)
    
    @property
    def active_connections(self):
        """Get the number of active connections"""
        return self.__active_connections
    
    @property
    def max_connections(self):
        """Get the maximum number of connections"""
        return self._max_connections
    
    @max_connections.setter
    def max_connections(self, value):
        """Set the maximum number of connections"""
        if not isinstance(value, int) or value <= 0:
            raise ValueError("max_connections must be a positive integer")
        self._max_connections = value


# Example 2: HTTP Request Handler
class HTTPRequest:
    """A simple HTTP request class"""
    
    def __init__(self, method, url, headers=None, body=None):
        self.method = method
        self.url = url
        self.headers = headers or {}
        self.body = body


class HTTPResponse:
    """A simple HTTP response class"""
    
    def __init__(self, status_code, body=None, headers=None):
        self.status_code = status_code
        self.body = body
        self.headers = headers or {}


class RequestHandler:
    """A class to handle HTTP requests"""
    
    def __init__(self, routes):
        self._routes = routes
        self.__middleware = []
    
    def handle_request(self, request):
        """Handle an HTTP request"""
        # Apply middleware
        modified_request = self.__apply_middleware(request)
        
        # Find the appropriate route handler
        handler = self._routes.get(modified_request.url)
        if not handler:
            return self.__create_not_found_response()
        
        # Call the handler with the request
        try:
            response = handler(modified_request)
            return response
        except Exception as e:
            return self.__create_error_response(str(e))
    
    def add_middleware(self, middleware_func):
        """Add a middleware function"""
        self.__middleware.append(middleware_func)
    
    def __apply_middleware(self, request):
        """Apply all middleware to the request"""
        modified_request = request
        for middleware in self.__middleware:
            modified_request = middleware(modified_request)
        return modified_request
    
    def __create_not_found_response(self):
        """Create a 404 Not Found response"""
        return HTTPResponse(
            status_code=404,
            body="Not Found",
            headers={"Content-Type": "text/plain"}
        )
    
    def __create_error_response(self, error_message):
        """Create a 500 Internal Server Error response"""
        return HTTPResponse(
            status_code=500,
            body=f"Internal Server Error: {error_message}",
            headers={"Content-Type": "text/plain"}
        )


# Example 3: ORM Model
class Model:
    """A base model class for a simple ORM"""
    
    def __init__(self, **kwargs):
        # Set initial attribute values
        for key, value in kwargs.items():
            setattr(self, key, value)
        
        # Private attributes for tracking state
        self.__dirty = False
        self.__deleted = False
    
    def save(self):
        """Save the model to the database"""
        if self.__deleted:
            raise Exception("Cannot save a deleted model")
        
        if hasattr(self, 'id') and self.id:
            self.__update()
        else:
            self.__insert()
        
        self.__dirty = False
    
    def delete(self):
        """Delete the model from the database"""
        if not hasattr(self, 'id') or not self.id:
            raise Exception("Cannot delete a model that hasn't been saved")
        
        # Simulate database delete
        print(f"DELETE FROM {self.__class__.__name__} WHERE id = {self.id}")
        
        self.__deleted = True
    
    def __insert(self):
        """Insert a new record into the database"""
        # In a real ORM, this would generate and execute an INSERT SQL statement
        print(f"INSERT INTO {self.__class__.__name__} ({', '.join(self.__get_fields())}) VALUES (...)")
        
        # Simulate getting an ID from the database
        if not hasattr(self, 'id') or not self.id:
            self.id = 1  # In a real scenario, this would be the ID from the database
    
    def __update(self):
        """Update an existing record in the database"""
        # In a real ORM, this would generate and execute an UPDATE SQL statement
        print(f"UPDATE {self.__class__.__name__} SET ... WHERE id = {self.id}")
    
    def __get_fields(self):
        """Get the model's fields"""
        return [key for key in self.__dict__ if not key.startswith('_')]
    
    @property
    def is_dirty(self):
        """Check if the model has unsaved changes"""
        return self.__dirty
    
    @property
    def is_deleted(self):
        """Check if the model has been deleted"""
        return self.__deleted
    
    def __setattr__(self, name, value):
        """Override setattr to track when attributes change"""
        # If this is an existing attribute, check if the value is changing
        if not name.startswith('_') and hasattr(self, name) and getattr(self, name) != value:
            self.__dirty = True
        
        # Set the attribute normally
        super().__setattr__(name, value)


# Create a specific model class
class User(Model):
    """User model for the ORM example"""
    
    def __init__(self, username, email, **kwargs):
        super().__init__(**kwargs)
        self.username = username
        self.email = email
    
    def __str__(self):
        if hasattr(self, 'id') and self.id:
            return f"User({self.id}, {self.username}, {self.email})"
        else:
            return f"User(unsaved, {self.username}, {self.email})"


# Test the examples
print("Example 1: Database Connection Pool")
pool = DatabaseConnectionPool(
    host="localhost",
    port=5432,
    user="postgres",
    password="secret",
    database="myapp",
    max_connections=3
)

# Get some connections
conn1 = pool.get_connection()
conn2 = pool.get_connection()
print(f"Active connections: {pool.active_connections}")
print(f"Available connections: {pool.available_connections}")

# Release a connection
pool.release_connection(conn1)
print(f"After releasing - Active: {pool.active_connections}, Available: {pool.available_connections}")

# Change max connections
pool.max_connections = 5
print(f"New max connections: {pool.max_connections}")

print("\nExample 2: HTTP Request Handler")
# Define some route handlers
def home_handler(request):
    return HTTPResponse(200, "Welcome to the home page", {"Content-Type": "text/html"})

def about_handler(request):
    return HTTPResponse(200, "About Us", {"Content-Type": "text/html"})

# Create a request handler
routes = {
    "/": home_handler,
    "/about": about_handler
}
handler = RequestHandler(routes)

# Add a middleware to log requests
def logging_middleware(request):
    print(f"Request: {request.method} {request.url}")
    return request

handler.add_middleware(logging_middleware)

# Handle some requests
print("\nHandling requests:")
response1 = handler.handle_request(HTTPRequest("GET", "/"))
print(f"Response: {response1.status_code}, {response1.body}")

response2 = handler.handle_request(HTTPRequest("GET", "/about"))
print(f"Response: {response2.status_code}, {response2.body}")

response3 = handler.handle_request(HTTPRequest("GET", "/nonexistent"))
print(f"Response: {response3.status_code}, {response3.body}")

print("\nExample 3: ORM Model")
# Create a user
user = User(username="john_doe", email="john@example.com")
print(f"New user: {user}")
print(f"Is dirty: {user.is_dirty}")

# Save the user
user.save()
print(f"After save - User: {user}")
print(f"Is dirty: {user.is_dirty}")

# Modify the user
user.email = "john.doe@example.com"
print(f"After modification - Is dirty: {user.is_dirty}")

# Save again
user.save()
print(f"After second save - Is dirty: {user.is_dirty}")

# Delete the user
user.delete()
print(f"Is deleted: {user.is_deleted}")

# Try to save a deleted user
try:
    user.save()
except Exception as e:
    print(f"Error: {e}")

Code Breakdown:

• Example 1: DatabaseConnectionPool
  • Demonstrates encapsulation in a connection pool implementation.
  • Uses protected attributes for configuration and private attributes for internal state.
  • Provides public methods for getting and releasing connections.
  • Uses properties to provide read access to internal state.
• Example 2: RequestHandler
  • Shows how encapsulation can be used in a web request handling context.
  • Uses private methods for internal operations like error handling and middleware application.
  • Provides a clean public interface for handling requests and adding middleware.
• Example 3: Model and User
  • Illustrates encapsulation in an ORM (Object-Relational Mapping) context.
  • Uses private attributes to track model state.
  • Overrides __setattr__ to track when attributes change.
  • Provides properties for accessing internal state.
  • Implements private methods for database operations.

These examples show how encapsulation is applied in real-world web development scenarios to create maintainable, flexible, and robust code.