Forem: Sakshi Agarwal

SOLID Principles: Building Robust Systems, Brick by Brick

Sakshi Agarwal — Wed, 26 Mar 2025 07:40:28 +0000

System design can feel like building a complex structure. Without a strong foundation, your system can become brittle, difficult to maintain, and prone to collapse under pressure. That's where the SOLID principles come in. These five principles, when applied correctly, act as the sturdy foundation for building robust, scalable, and maintainable software systems.

Let's break them down in the easiest way possible:
Think of building a house. You want it to be strong, easy to fix, and easy to add to later. That’s what SOLID principles do for software.

1. One Thing Per Room-Single Responsibility Principle (SRP)

What it means:
Each room in your house has one main job. The kitchen is for cooking, the bedroom is for sleeping, etc.
In code:
Each “part” of your software should do one thing well. Don’t try to make your kitchen also be the garage.
Why it's good:
If the kitchen sink breaks, you don't have to rebuild the whole house. You just fix the kitchen.

# House Analogy: Kitchen does cooking, bedroom does sleeping.
# Code Example:

class Kitchen:
    def cook_meal(self, meal):
        print(f"Cooking {meal} in the kitchen.")

class Bedroom:
    def sleep(self):
        print("Sleeping in the bedroom.")

# Bad Example (mixing responsibilities):
# class MultiPurposeRoom:
#     def cook_meal(self, meal):
#         # ...
#     def sleep(self):
#         # ...

my_kitchen = Kitchen()
my_bedroom = Bedroom()

my_kitchen.cook_meal("dinner")
my_bedroom.sleep()

2. Adding Rooms Without Tearing Down Walls (Open/Closed Principle)

What it means:
You should be able to add a new room (like a sunroom) without having to knock down existing walls.
In code:
You should be able to add new features to your software without changing the old code.
Why it's good:
You don't accidentally break the living room while building the sunroom.

class HouseRoom:
    def area(self):
        pass  # Base room, can't calculate area alone

class Bedroom(HouseRoom):
    def __init__(self, length, width):
        self.length = length
        self.width = width

    def area(self):
        return self.length * self.width

class Sunroom(HouseRoom):
    def __init__(self, length, width):
        self.length = length
        self.width = width

    def area(self):
        return self.length * self.width

def calculate_room_area(room):
    print(f"Room area: {room.area()}")

my_bedroom = Bedroom(10, 12)
my_sunroom = Sunroom(8, 8)
calculate_room_area(my_bedroom)
calculate_room_area(my_sunroom)

# Bad Example (modifying existing code to add new room types):
# def calculate_room_area(room, room_type):
#     if room_type == "bedroom":
#         print(f"Bedroom area: {room.length * room.width}")
#     elif room_type == "sunroom":
#         print(f"Sunroom area: {room.length * room.width}")
#     # ... (would need to modify this function for every new room type)

3.Using Interchangeable Parts (Liskov Substitution Principle)

What it means:
Think of standard sized building materials. If you design a door frame for a "standard door," you should be able to put in any door that fits that standard size, whether it's a wooden door, a metal door, or a glass door. All these doors should "fit" where a "standard door" is expected
In code:
If you have a general "shape" for something (like a "door"), all the specific types of that thing (like "wooden door," "metal door") should work in the same way wherever you use the general "shape."
Why it's good:
It means you can easily swap out parts without having to rebuild the whole structure. If you need a more secure door, you can just replace the wooden one with a metal one, and it should still fit the same frame.

class Door:
    def fit(self, frame_width, frame_height):
        pass

class WoodenDoor(Door):
    def fit(self, frame_width, frame_height):
        print(f"Wooden door fitting in {frame_width}x{frame_height} frame.")

class MetalDoor(Door):
    def fit(self, frame_width, frame_height):
        print(f"Metal door fitting in {frame_width}x{frame_height} frame.")

def install_door(door, frame_width, frame_height):
    door.fit(frame_width, frame_height)

my_wooden_door = WoodenDoor()
my_metal_door = MetalDoor()

install_door(my_wooden_door, 36, 80)
install_door(my_metal_door, 36, 80)

# Bad Example (violating LSP):
# class Wall:
#     def fit(self, frame_width, frame_height):
#         raise Exception("Walls do not fit in door frames.") # Wall doesnt work where door is expected.
#
# my_wall = Wall()
# install_door(my_wall, 36, 80) # This would fail.

4. Only the Tools You Need (Interface Segregation)

What it means:
If you’re building a house, you don’t carry around every tool. You bring the hammer for nails, the saw for wood.
In code:
If a part of your software only needs a few abilities, don’t give it all the abilities. Give it just the ones it needs.
Why it's good:
It keeps things simple. The hammer doesn’t get in the way when you’re using the saw.

class HammerTool:
    def hammer_nails(self):
        print("Hammering nails.")

class SawTool:
    def cut_wood(self):
        print("Cutting wood.")

class SimpleBuilder:
    def use_hammer(self, hammer):
        hammer.hammer_nails()

class Carpenter:
    def use_hammer(self, hammer):
        hammer.hammer_nails()
    def use_saw(self, saw):
        saw.cut_wood()

my_hammer = HammerTool()
my_saw = SawTool()

builder = SimpleBuilder()
builder.use_hammer(my_hammer)

carpenter = Carpenter()
carpenter.use_hammer(my_hammer)
carpenter.use_saw(my_saw)

# Bad Example (large interface, not all builders need all tools):
# class BuilderTool:
#     def hammer_nails(self):
#         pass
#     def cut_wood(self):
#         pass
#
# class SimpleBuilder:
#     def use_tool(self, tool: BuilderTool):
#         tool.hammer_nails()
#
# class Carpenter:
#     def use_tool(self, tool: BuilderTool):
#         tool.hammer_nails()
#         tool.cut_wood()

5. Don’t Care Who Made the Pipes, Just That They Work (Dependency Inversion)

What it means:
You don’t care who made the pipes in your house, just that they carry water.
In code:
Your software shouldn’t depend on specific “parts.” It should depend on general “ways of doing things.”
Why it's good:
If the pipe company goes out of business, you can easily switch to another brand of pipes. The water still flows.

class WaterSource:
    def supply_water(self):
        pass

class Pipe:
    def __init__(self, source: WaterSource):
        self.source = source

    def deliver_water(self):
        self.source.supply_water()
        print("Water delivered through pipe.")

class CityWater(WaterSource):
    def supply_water(self):
        print("City water supply.")

class WellWater(WaterSource):
    def supply_water(self):
        print("Well water supply.")

city_water = CityWater()
well_water = WellWater()

my_pipe = Pipe(city_water)
my_pipe.deliver_water()

my_pipe = Pipe(well_water)
my_pipe.deliver_water()

# Bad Example (depending on concrete implementations):
# class Pipe:
#     def __init__(self, city_water): # Hard dependency on CityWater
#         self.city_water = city_water
#
#     def deliver_water(self):
#         self.city_water.supply_water()
#         print("Water delivered through pipe.")
#
# my_pipe = Pipe(CityWater()) # Would break if we needed to use WellWater.

In Simple Words:
SOLID helps you build software that:

Is easy to understand.
Is easy to change.
Is less likely to break when you change it.
Is easy to reuse parts of.
It's like building a house that’s well-organized, easy to fix, and easy to add to later. You would never build a house where changing the kitchen broke the bathroom, and neither should you build software that way.

Efficiently Deleting Multiple Lakhs of Records from DynamoDB Using Lambda: A Comprehensive Guide

Sakshi Agarwal — Sun, 29 Sep 2024 17:39:02 +0000

DynamoDB, a highly scalable NoSQL database service offered by AWS, is designed for high throughput and low latency. However, deleting a massive number of records at once can be a performance-intensive operation. This blog post will guide you through the process of efficiently deleting multiple lakhs of records from DynamoDB using a Lambda function, incorporating best practices and addressing common challenges.

Key Considerations
Before diving into the code, it's essential to consider the following:

Partition Key: DynamoDB uses a partition key to distribute data across partitions. Ensure that the partition key values are evenly distributed to avoid hot partitions.
Batch Write Operations: DynamoDB supports batch write operations, which can significantly improve performance by allowing you to delete multiple items in a single request.

Throttling: Be mindful of DynamoDB's throttling limits. Exceeding these limits can result in errors and delays.

Consistency: If consistency is crucial, consider using DynamoDB's strong consistency mode. However, this may impact performance.

Filter Expression: Use a filter expression to selectively delete records based on specific criteria, reducing the number of items to be processed.

Lambda Function Code

Here's a Python Lambda function that demonstrates how to efficiently delete multiple lakhs of records from DynamoDB using batch write operations, filter expressions, and last evaluated key handling:

import boto3
import time

dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('your_table_name')

def delete_records(records, last_evaluated_key):
    with table.batch_writer() as batch:
        for record in records:
            batch.delete_item(Key={'partition_key': record['partition_key'], 'sort_key': record['sort_key']})

    return batch.response.get('LastEvaluatedKey')

def handler(event, context):
    filter_expression = "attribute_name = :value"
    expression_attribute_values = {':value': "your_filter_value"}

    response = table.scan(
        FilterExpression=filter_expression,
        ExpressionAttributeValues=expression_attribute_values
    )

    records = response['Items']
    last_evaluated_key = response.get('LastEvaluatedKey')

    while last_evaluated_key:
        response = table.scan(
            FilterExpression=filter_expression,
            ExpressionAttributeValues=expression_attribute_values,
            ExclusiveStartKey=last_evaluated_key   

        )

        records.extend(response['Items'])
        last_evaluated_key = response.get('LastEvaluatedKey')

    # Divide the records into batches to avoid throttling
    batch_size = 25
    batches = [records[i:i+batch_size] for i in range(0, len(records), batch_size)]

    for batch in batches:
        last_evaluated_key = delete_records(batch, last_evaluated_key)
        # Introduce a delay to avoid throttling
        time.sleep(1)

    return {'statusCode': 200, 'body': 'Records deleted successfully'}

Explanation:

Filter Expression: The filter_expression and expression_attribute_values parameters allow you to selectively delete records based on specific criteria, reducing the number of items to be processed.
Last Evaluated Key: The last_evaluated_key is used to efficiently handle large datasets by retrieving items in batches and continuing from the last processed item.
Batch Processing: The records are divided into batches to avoid throttling and improve performance. The delete_records function now returns the LastEvaluatedKey to be used in subsequent scans.
Looping: The code uses a loop to continue scanning and deleting records until there are no more items to process.

Additional Considerations

Error Handling: Implement proper error handling mechanisms to catch exceptions and log errors for troubleshooting.
Performance Optimization: If you're dealing with an extremely large number of records, consider using DynamoDB's global secondary indexes or partitioning strategies to improve performance.
Parallel Processing: For even faster deletion, explore using multiple Lambda functions or a distributed system to process the records in parallel.
Cost Optimization: If cost is a concern, carefully evaluate the number of Lambda invocations and the amount of data transferred to optimize your costs.
By following these guidelines and customizing the code to your specific use case, you can effectively delete multiple lakhs of records from DynamoDB using a Lambda function, ensuring efficient, scalable, and cost-effective operations.

Building Scalable GraphQL APIs with AWS AppSync: A Deep Dive into Operations

Sakshi Agarwal — Tue, 12 Sep 2023 10:21:59 +0000

Before moving on to App Sync and its core operations, let's understand GraphQL in simpler terms.
GraphQL is a modern API query language that enables clients to request specific data from a server, eliminating over-fetching and under-fetching of information. It uses a schema to define data types and operations, giving clients control over the data they receive. This flexibility and efficiency make it a powerful alternative to traditional REST APIs.

What is AWS AppSync?

AWS AppSync is a managed service that allows you to easily develop GraphQL APIs by connecting to various data sources. It abstracts the complexities of managing the infrastructure and provides real-time data synchronization and offline access to your applications.

In this technical blog post, we'll delve into AWS AppSync and explore its core operations: Query, Mutation, and Subscription.

Operation 1: Query
Purpose: Query operations are used for fetching data from your GraphQL API. Think of them as read-only operations.

query {
  getTodo(id: "123") {
    id
    title
    completed
  }
}

Explanation:

In this query, we're requesting specific fields (id, title, and completed) for a todo item with the ID "123". AWS AppSync retrieves this data from your data source, and the response contains the requested information.

Operation 2: Mutation
Purpose: Mutation operations modify data in your GraphQL API. You can use mutations to create, update, or delete records.

mutation {
  createTodo(input: {
    title: "Buy groceries",
    completed: false
  }) {
    id
    title
    completed
  }
}

Explanation:

This mutation creates a new todo item with the specified title ("Buy groceries") and completion status (false). The response provides details about the newly created todo item, including its ID, title, and completion status.

Operation 3: Subscription
Purpose: Subscription operations enable real-time updates. Clients can subscribe to specific events, and the server pushes updates to subscribed clients when those events occur.

subscription {
  onTodoUpdated {
    id
    title
    completed
  }
}

Explanation:

This operation allows clients to receive real-time updates when todo items are updated. It listens for changes to the specified fields (id, title, and completed) and sends immediate updates to subscribed clients. This feature is crucial for building responsive applications with live data.

The Advantages of AWS AppSync Operations

Simplicity: AWS AppSync abstracts the complex setup of GraphQL servers, making it easy to get started with API development.

Real-Time Data: Subscriptions in AWS AppSync simplify the implementation of real-time features, such as chat applications, live dashboards, and collaborative editing.

Security: AWS AppSync seamlessly integrates with AWS Cognito and other authentication providers, ensuring secure access control to your data.

Flexibility: With AppSync, you can connect to various data sources, including databases like DynamoDB and RESTful APIs.

Scalability: AWS AppSync automatically scales to handle high loads, eliminating the need for manual infrastructure provisioning.

In Conclusion
AWS AppSync's operations make building GraphQL APIs straightforward. Use Queries for reading data, Mutations for modifying data, and Subscriptions for real-time updates. Simplify API development, focus on user experiences, and leverage AWS AppSync for your next project.

Building a Serverless User Details Fetching API with AWS Lambda, DynamoDB, OOP Principles and Powertools

Sakshi Agarwal — Thu, 25 May 2023 07:06:13 +0000

Introduction

In the world of serverless computing, AWS Lambda has gained popularity for its scalability and cost efficiency. In this blog post, we will explore how to build a Lambda function that fetches user details from DynamoDB using object-oriented programming (OOP) principles. To enhance our function with advanced observability and operational best practices, we will leverage the power of the AWS Lambda Powertools library.

Prerequisites:
To follow along with this tutorial, you should have a basic understanding of AWS Lambda, DynamoDB, and Python programming. You should also have an AWS account and the necessary permissions to create Lambda functions and access DynamoDB.

Here's an example of a Lambda function written using object-oriented programming (OOP) principles to fetch user details from DynamoDB:

from aws_lambda_powertools import Logger, Tracer, Validator
from aws_lambda_powertools.utilities.data_classes import APIGatewayProxyEvent

import boto3

logger = Logger()
tracer = Tracer()
validator = Validator()

class UserFetcher:
    def __init__(self, table_name):
        self.dynamodb = boto3.resource('dynamodb')
        self.table = self.dynamodb.Table(table_name)

    def fetch_user_details(self, user_id):
        response = self.table.get_item(Key={'user_id': user_id})
        user_details = response.get('Item')
        return user_details

@tracer.capture_lambda_handler
@logger.inject_lambda_context
def lambda_handler(event: APIGatewayProxyEvent, context):
    # Validate input using powertools.Validator
    validator.validate(event, "user_id")
    user_id = event.path_parameters.get('user_id')

    # Initialize UserFetcher object
    user_fetcher = UserFetcher('UserTable')

    try:
        # Fetch user details from DynamoDB
        user_details = user_fetcher.fetch_user_details(user_id)
        logger.info(f"Fetched user details: {user_details}")

        return {
            "statusCode": 200,
            "body": user_details
        }
    except Exception as e:
        logger.error(f"Error fetching user details: {str(e)}")
        return {
            "statusCode": 500,
            "body": "Error fetching user details"
        }

Step 1: Setting Up the Environment
Before we begin writing code, let's set up our environment. Create a new Lambda function using the AWS Management Console or the AWS CLI. Choose a Python runtime, and configure the function with appropriate permissions to access DynamoDB. Install the AWS Lambda Powertools library by adding it to your function's deployment package.

Step 2: Writing the UserFetcher Class
To encapsulate the logic for fetching user details from DynamoDB, we will create a UserFetcher class using OOP principles. This class will have a constructor (init) method to initialize the DynamoDB resource and table, and a fetch_user_details method that takes a user_id as input and retrieves the corresponding user details from DynamoDB using the get_item API.

Step 3: Implementing Input Validation and Logging with Powertools
To enhance our function's robustness and visibility, we will use the AWS Lambda Powertools library. Within the lambda_handler function, we will use the powertools.utilities.validator module to validate the input and raise an exception if it's missing. Additionally, we will leverage powertools.Logger to log informative messages and handle any exceptions that occur during the execution of our Lambda function.

Step 4: Testing and Deploying the Lambda Function
Once we have written the code for our Lambda function, we can test it locally using a tool like the AWS SAM CLI or by creating a test event in the AWS Lambda console. Ensure that your DynamoDB table exists and has the required user data. Deploy the Lambda function to your AWS account using the deployment mechanism of your choice.

Conclusion:
In this blog post, we learned how to build a serverless API using AWS Lambda, DynamoDB, and OOP principles. We enhanced our Lambda function with input validation and logging using the AWS Lambda Powertools library. By following the step-by-step guide and leveraging Powertools, we created a robust and observable solution for fetching user details. This approach allows us to build scalable and reliable serverless applications while adhering to best practices and leveraging advanced tools.

JSON Web Tokens(JWT) For Authentication and Authorization in AWS

Sakshi Agarwal — Mon, 17 Apr 2023 16:30:43 +0000

JSON Web Tokens (JWTs) are a compact type of access token that is used to securely transmit information between parties. In the context of authentication, a JWT can be used as a secure and stateless way to authenticate users and authorize access to resources. It can be verified locally by the client without the need to call the third-party service (such as an OAuth provider) every time the client wants to access a protected resource.

JWTs can be verified locally because they are self-contained and contain all the necessary information to verify their authenticity. A JWT consists of three parts: a header, a payload, and a signature.

The header contains information about the type of token and the algorithm used to sign it.
The payload contains the claims or assertions about the user or entity that the token represents.
The signature is created by combining the header, payload, and a secret key that is only known to the server. The signature can be used to verify the authenticity of the JWT.

When a client sends a JWT to a server, the server can verify the authenticity of the token by decoding the token, checking the signature using the secret key, and verifying that the claims in the payload are valid. Since the secret key is only known to the server, this process ensures that only tokens issued by the server are accepted, and that the claims in the token have not been tampered with.

In contrast to OAuth, which relies on a centralized authorization server to manage access tokens, JWTs can be verified locally without the need for a third-party service. This makes JWTs a good option for applications that need to authenticate and authorize users without relying on external services.

Here's how you can use JWTs for authentication and authorization:

The client (such as a web or mobile app) sends a request to your server with a username and password.
- Your server authenticates the user and generates a JWT that includes the user's ID and any relevant permissions or roles.
- Your server sends the JWT back to the client in the response.
- The client stores the JWT (usually in local storage or a cookie) and sends it in the Authorization header of subsequent requests to your server.
- Your server verifies the JWT by checking the signature and decoding the claims (such as the user ID and permissions).
- If the JWT is valid, your server allows the request and returns the requested resource or performs the requested action.
- If the JWT is invalid or expired, your server returns an error response. _

To use JWT (JSON Web Token) in AWS using Python, you can follow these steps:

1.Create a JWT token using PyJWT:



import jwt

def lambda_handler(event, context):
    # Generate a JWT with a payload containing the user's data
    token = jwt.encode({'username': 'johndoe', 'role': 'admin'}, 'my-secret-key', algorithm='HS256')

    return {
        'statusCode': 200,
        'body': token,
    }

In this example, the payload is a dictionary that includes the user ID and permissions. The secret key is a string that is used to sign the JWT. The jwt.encode() function encodes the payload and signs it with the secret key using the HS256 algorithm.

2.- Verify the JWT in API Gateway:
Make an API request to an AWS service (e.g., AWS Lambda) and include the JWT token in the request headers:



import requests

url = 'https://your-lambda-function-endpoint.amazonaws.com'
headers = {'Authorization': f'Bearer {token.decode()}'}

response = requests.get(url, headers=headers)

In the above example, we make an HTTP GET request to an AWS Lambda function using the requests library. We include the JWT token in the request headers using the Authorization field with the Bearer prefix.

Note that the token variable needs to be decoded to a string before it can be included in the headers.

That's it! You should now be able to use JWT in AWS using Python.

AWS SES with Lambda

Sakshi Agarwal — Wed, 15 Mar 2023 10:55:27 +0000

Amazon SES (Simple Email Service) is a simple service that can be used by small and large industries to send marketing, transaction and notification emails as the cost is less and it is reliable. SES can be easily integrated to an existing application with the help of SMTP or AWS SDK.

Why SES ?

• Building an in house email solutions is a complex and expensive challenge for a business due to the infrastructure demands, network assembly and security involved. Also some proper third-
party email solutions require huge up-front prices. SES eliminates these complexities.
• It have the benefit of a refined email infrastructure that amazon has used for its large consumer base.
• With SES there are not any direct prices and no minimum commitments. If we use SES in an Amazon EC2 instance, the first 62,000 emails sent are free and a very low price for emails sent
thereafter.
• Also SES can be integrated with other AWS services to provide seamless email solutions to the end users. You pay as you go, and you pay just for what you utilize.

SES Configuration

SES can be configured to be used in two ways.
• Use SES as outbound email server
• Using existing server and configure it to send the mails through Amazon SES.

SES with AWS Lambda

To send Emails from a Lambda function using SES we need to configure the following.

• Required IAM Lambda permissions for the API call.

ses:SendEmail – Email is immediately queued for sending after composing
ses:SendRawEmail – This is suitable in cases where we need to add attachments or HTML content.

These policies can be attached to an IAM role that would be assigned to our Lambda function.

• Verified SES identity.

To use SES it is required to verify sender email addresses to confirm that the sender owns them and to prevent unauthorized use.
We must verify each identity that we use. From, Source, Sender and Return address
Email addresses are case sensitive
If both email address and the domain the mail ID belongs is verified, the email address settings override the domain settings.
Verification status is different for each region. If we want to send mails from the same identity from different regions it has to be verified in all regions.

Following is a simple Lambda example function using SES.

import boto3
client = boto3.client(
    'ses',
    region_name=region,
    aws_access_key_id='aws_access_key_string',
    aws_secret_access_key='aws_secret_key_string'
)
response = client.send_email(
    Destination={
        'ToAddresses': ['recipient1@domain.com', 'recipient2@domain.com'],
    },
    Message={
        'Body': {
            'Text': {
                'Charset': 'UTF-8',
                'Data': 'email body string',
            },
        },
        'Subject': {
            'Charset': 'UTF-8',
            'Data': 'email subject string',
        },
    },
    Source='sender.email@domain.com',
)

Benefits of SES

High Deliverability:
Features like content filtering and reputation dashboard ensure high number of emails are delivered. Higher the sender’s reputation more the number of emails sent from that address.

Cost-Effective:
No upfront or fixed costs and the cost depends on our monthly usage of this service.

Configurable:
SES can be configured with other AWS services like EC2, ECS and Cloudwatch to get detailed analysis of the mail metrics and also get notifications using SNS.

Use Cases

Transactional:
SES can be used to send automated mails for order confirmation, delivery, and subscription related information.

Notifications:
Notifications on application health and metrics and send user notifications when an event is triggered.

Incoming Mails:
SES can be used to receive mails and programmatically route the same to a S3 bucket.

Marketing:
Promotions, offers and new product details can be sent in high quality emails using SES.