DEV Community: Idris Rampurawala

Building a Scalable API Event Logger using Pub/Sub, and BigQuery

Idris Rampurawala — Mon, 03 Nov 2025 13:44:26 +0000

API logging is one of the most underrated aspects of backend engineering. We often rely on structured logs for debugging, but when it comes to analytics, user behavior tracking, or performance insights, traditional logs begin to show their limitations.

That’s where event-based API logging comes in.

In this post, I’ll walk you through building a simple but scalable API event logger using Go, Google Pub/Sub, and BigQuery. We’ll also see how this approach differs from traditional GCP logs and why it might be the better choice for data-driven teams.

☑️ Prerequisite

Before we begin, we are assuming that you have some knowledge of Middlewares, Google Cloud Pub/Sub, and BigQuery (data warehouse)
Though this approach can be implemented with any language/framework, I have created a working Go application project

Let's roll in!🍟

📇 Overview

Modern APIs generate a continuous stream of data about how users interact with your system — but traditional logging methods often treat this as an afterthought, providing debugging information rather than actionable events. Event-based API logging flips that model: every API interaction becomes a structured, analyzable event.

Before diving into how this logger works with Pub/Sub and BigQuery, let’s first understand what event-based API logging means and why conventional logging systems can fall short for analytics and long-term insights.

What does "event-based API logging" mean?

Event-based API logging treats each API request (and response) as a discrete event, capturing key attributes such as method, URL, user ID, timestamp, request/response payloads, and more. Unlike traditional logs that mix system messages, debug lines, and errors, an event-based approach focuses on meaningful business or technical interactions (e.g., "user 123 hit endpoint /v1/items", "response code 201").

By publishing these events to a message broker (such as Google Cloud Pub/Sub) and then storing them in an analytics store (like BigQuery), we enable downstream systems — dashboards, ML jobs, user-behavior pipelines — to unsubscribe, transform, and analyze without modifying the producer.

What are some limitations of "old-style" logging?

While tools like Google Cloud Logging (formerly Stackdriver) are great for operations, they impose constraints when we want long-term analytics or behavioural metrics. For instance, the default retention period for log buckets is 30 days, after which entries are deleted unless we change it.

On the querying side, Logs Explorer is purpose-built for debugging rather than large-scale analytics; we don't get the same SQL flexibility or performance as a dedicated data warehouse. In short, for one-off troubleshooting, it works; for streaming user-behaviour metrics, it often doesn't.

Now let's dive deep into our implementation 👇

🧠 API Logging Flow

The general idea is to capture all API calls using a middleware and send a Pub/Sub event with all the information. The subscriber then consumes this Pub/Sub message and creates a record in BigQuery.

Here are the steps involved:

Client makes an API call
API Server (Go app in this example) processes the request and logs details like method, URL, user_id, request/response JSON
The Logging Middleware publishes this log as an event to Pub/Sub
A GCP BigQuery subscriber automatically consumes these events and stores them in a BigQuery table
We can now run SQL queries on your API activity!

Now that we know the flow, let's discuss the Middleware implementation

⚙️ The Logging Middleware

The APILogEvent struct

// APILogEvent represents an API request/response log event
type APILogEvent struct {
    RequestID    null.String `json:"request_id"`
    Service      string      `json:"service"`
    URL          string      `json:"url"`
    Method       string      `json:"method"`
    ResponseCode int         `json:"response_code"`
    ResponseBody null.String `json:"response_body,omitempty"`
    RequestBody  null.String `json:"request_body,omitempty"`
    UserID       null.String `json:"user_id,omitempty"`
    Duration     float64     `json:"duration"`
    Version      string      `json:"version"`
    Name         string      `json:"name"`
    CreatedAt    time.Time   `json:"created_at"`
}

Service - represents the API service name, so we can use it across multiple API services
UserID - represents the caller of the API

The Middleware

package middleware

import (
    "bytes"
    "context"
    "fmt"
    "io"
    "log"
    "net/http"
    "strings"
    "time"

    "api-pubsub-logger/internal/pubsub"
    "api-pubsub-logger/internal/utils"
    "api-pubsub-logger/pkg/logger"

    "github.com/gorilla/mux"
    "gopkg.in/guregu/null.v3"
)

// Routes to skip from logging (e.g., health checks)
var skipRoutes = map[string]struct{}{
    "GET::/health": {},
}

// responseRecorder is a wrapper for http.ResponseWriter to capture response data
type responseRecorder struct {
    http.ResponseWriter
    body       *bytes.Buffer
    statusCode int
}

func (rw *responseRecorder) Write(b []byte) (int, error) {
    rw.body.Write(b)
    return rw.ResponseWriter.Write(b)
}

func (rw *responseRecorder) WriteHeader(statusCode int) {
    rw.statusCode = statusCode
    rw.ResponseWriter.WriteHeader(statusCode)
}

// LoggingMiddleware logs HTTP requests and responses to Pub/Sub
func LoggingMiddleware(pubsubClient pubsub.Publisher, serviceName string) func(http.Handler) http.Handler {
    return func(next http.Handler) http.Handler {
        return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
            // Check if the request URL path should be skipped
            routeKey := fmt.Sprintf("%s::%s", strings.ToUpper(r.Method), r.URL.Path)
            if _, skip := skipRoutes[routeKey]; skip {
                next.ServeHTTP(w, r)
                return
            }

            startTime := time.Now()

            // Read request body
            var requestBody []byte
            if r.Body != nil {
                requestBody, _ = io.ReadAll(r.Body)
                r.Body = io.NopCloser(bytes.NewBuffer(requestBody)) // Restore the request body
            }

            // Create response recorder to capture response
            recorder := &responseRecorder{
                ResponseWriter: w,
                body:           &bytes.Buffer{},
                statusCode:     http.StatusOK,
            }

            // Call the next handler
            next.ServeHTTP(recorder, r)

            // Extract context values
            ctx := r.Context()
            requestID := utils.GetRequestID(ctx)
            userID := utils.GetUserID(ctx)

            // Mask sensitive data in request and response bodies
            maskedRequestBody := string(utils.MaskSensitiveData(requestBody))
            maskedResponseBody := string(utils.MaskSensitiveData(recorder.body.Bytes()))

            // Extract route version and name from mux router
            routeName, routeVersion := extractRouteVersionAndName(mux.CurrentRoute(r))

            // Create API log event
            logData := logger.APILogEvent{
                RequestID:    null.NewString(requestID, len(requestID) > 0),
                Service:      serviceName,
                Method:       r.Method,
                URL:          r.URL.String(),
                RequestBody:  null.NewString(maskedRequestBody, len(maskedRequestBody) > 0),
                ResponseBody: null.NewString(maskedResponseBody, len(maskedResponseBody) > 0),
                ResponseCode: recorder.statusCode,
                UserID:       null.NewString(userID, len(userID) > 0),
                Version:      routeVersion,
                Name:         routeName,
                CreatedAt:    startTime,
                Duration:     time.Since(startTime).Seconds(),
            }

            // Publish to Pub/Sub asynchronously using background context
            // We use context.Background() instead of the request context because
            // the request context gets canceled when the HTTP response is sent,
            // but we want the publishing to complete independently
            go sendToPubSub(context.Background(), pubsubClient, logData)
        })
    }
}

// extractRouteVersionAndName extracts the name and version from the mux route
func extractRouteVersionAndName(route *mux.Route) (string, string) {
    var name, version string
    if route != nil {
        name = route.GetName()
        pathTemplate, _ := route.GetPathTemplate()
        parts := strings.Split(strings.Trim(pathTemplate, "/"), "/")
        if len(parts) > 0 && strings.HasPrefix(parts[0], "v") {
            version = parts[0]
        }
    }
    return name, version
}

// sendToPubSub publishes log data to Pub/Sub
func sendToPubSub(ctx context.Context, client pubsub.Publisher, logData logger.APILogEvent) {
    if err := client.PublishAPILogEvent(ctx, logData); err != nil {
        log.Printf("Failed to publish API log event: %v", err)
    }
}

The router can be set up as

package http

import (
    "net/http"

    "api-pubsub-logger/internal/http/handlers"
    "api-pubsub-logger/internal/http/middleware"

    "github.com/gorilla/mux"
)

// Handler mounts all the handlers at the appropriate routes and adds any required middleware
func (h *Handler) Handler() http.Handler {
    r := mux.NewRouter()

    // Apply global middleware
    r.Use(middleware.RequestIDMiddleware)
    r.Use(middleware.UserIDMiddleware)
    r.Use(middleware.LoggingMiddleware(h.PubSubClient, h.ServiceName))

    // Health check endpoint (not logged due to skip in middleware)
    r.Methods("GET").Path("/health").Name("health").HandlerFunc(handlers.HealthCheck)

    h.router = r
    return r
}

Let's understand a bit deeper what's happening 🙇

The Pub/Sub client is set up in main.go and part of the HTTP Handler struct, which the Middleware then accesses
We allow skipping logging for the API/routes using skipRoutes map
To ensure sensitive information is not exposed, we define a map sensitiveKeys defined in utils/mask.go that lists fields to mask
The logs are pushed to Pub/Sub in the background, so the API latency remains unaffected

🧩 Integration with BigQuery

Once Pub/Sub receives the log events, we can configure a direct BigQuery subscription in GCP.
👉 Stream Pub/Sub messages to BigQuery

Each message is appended as a new row in our api_logs table — matching the APILogEvent structure. BigQuery automatically maps these fields when the Pub/Sub topic is linked to a dataset.

That's it! 🎯

🤔 Why Not Just Use GCP Structured Logs?

GCP’s structured logs are powerful but not ideal for data analytics use cases.

Feature	Cloud Logging	Pub/Sub + BigQuery Logger
Query Language	Logs Explorer (limited SQL)	Full BigQuery SQL
Query performance	Slow for analytics	Fast, SQL-based via BigQuery
Custom schema	Limited	Fully customizable
Integration	GCP only	Can export anywhere

📊 What We Can Do with the Data

Once in BigQuery, we can easily compute metrics such as:

API response time in last 7 days

SELECT name, version, COUNT(*) api_count, MIN(duration) min_duration, AVG(duration) avg_duration,  MAX(duration) max_duration
FROM `api_logs` 
WHERE 
  created_at >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 7 DAY)
GROUP BY name, version
ORDER BY avg_duration DESC;

🚀 Benefits

✅ Asynchronous Logging — doesn’t block API latency.
✅ Scalable — Pub/Sub handles millions of events.
✅ Analytics Ready — BigQuery can query directly.
✅ Customizable Schema — include app version, route, or user agent.
✅ Cross-Language Ready — can be replicated in Node.js, Python, etc.

📌 You can check out my GitHub repository for a complete working example of this approach 👇

idris-rampurawala / api-pubsub-logger

Demo Go API with middleware that logs requests/responses to Google Cloud Pub/Sub. Features versioned routes, sensitive data masking, request tracking, and comprehensive unit tests.

API Pub/Sub Logger

A demonstration project showcasing how to log API requests and responses to Google Cloud Pub/Sub for later analysis and debugging.

Blog Post

For a detailed explanation of this implementation, read the full blog post Building a Scalable API Event Logger using Pub/Sub, and BigQuery

Overview

This project demonstrates a pattern for logging all API requests and responses to Google Cloud Pub/Sub. The middleware captures:

Request/Response bodies (with sensitive data masking)
HTTP method and URL
Response status codes
Request duration
Request IDs for tracing
User IDs from headers

All logs are published to a Pub/Sub topic which can then be consumed by subscribers to store in BigQuery or other analytics platforms.

Features

Middleware-based logging: Automatic logging of all API requests
Sensitive data masking: Automatically redacts email, phone numbers, and other sensitive fields
Request ID tracking: Unique ID for each request for distributed tracing
API versioning…

View on GitHub

📑 References

See ya! until my next post 😋

Implementing a Redis-Based Task Queue with Configurable Concurrency

Idris Rampurawala — Thu, 30 Jan 2025 21:52:32 +0000

In the realm of modern application development, efficiently managing background tasks is crucial. Redis, renowned for its versatile data structures and high performance, serves as an excellent foundation for building a task queue system. In this post, we'll delve into creating a Redis-based queue with configurable concurrency.

Working in various startups I have had the liberty to solve complex problems through quick but creative solutions. In this post, we will discuss one situation where I had to design a task queue that was easy to configure, manage and cost-effective. Task queues let applications perform work, called tasks, asynchronously outside of a user request. If an app needs to execute work in the background, it adds tasks to task queues. The tasks are executed later, by worker services. Now that we discussed what are task queues, let's dive into the implementation.

☑️ Prerequisite

Before we begin, we are assuming that you have some knowledge of Redis and Celery
We will be using Redis and Celery setup in Python forked from my other project Flask Boilerplate

Let's roll in!🍟

Overview

The task queue system leverages Redis lists to store tasks and processes them asynchronously. We will create two celery functions, which will add tasks to the queue and process them in sequence as per the concurrency level.

1. `queue_task_executor`

This function does two things:

adds the task to the queue
processes the task when called by callback (2nd function)

@celery.task(name='core.tasks.queue_task_executor')
def queue_task_executor(**kwargs):
    # fetch the task id (request_id)
    request_id = kwargs.get('request_id')
    is_called_by_callback = kwargs.get('is_called_by_callback', False)
    if not is_called_by_callback:
        # push this task to the tail of the queue
        logger.info(f'pushing id: {request_id} to queue')
        redis_client.rpush(TASK_EXEC_LIST_KEY, request_id)
        # calling callback which will pick it up for execution
        queue_task_executor_callback.apply_async(kwargs={})
        return
    # execute the task here
    # time.sleep(5)
    logger.info(f'task id:{request_id} execution completed successfully!')

It simply checks if the function is called by a callback to execute the task otherwise just adds the task to the tail of the Redis list (queue). The statement time.sleep(5) is just a test line that can be replaced with the task call/logic. We can simply call appropriate task functions (to execute) based on the arguments (request_id) being passed as metadata to queue the task.

2. `queue_task_executor_callback`

This callback function acts as a brain of the system. It performs the following:

Pick the next bunch of tasks to run in parallel from the Redis list
Uses Celery Groups to queue the task with a self-callback

@celery.task(name='core.tasks.queue_task_executor_callback')
def queue_task_executor_callback(**kwargs):
    logger.debug(f'queue_task_executor_callback called {kwargs}')
    if kwargs and kwargs.get('pick_next'):  # only delete this key if called via completing tasks
        redis_client.delete(TASK_EXECUTOR_KEY)
    if redis_client.setnx(TASK_EXECUTOR_KEY, 1):
        # retrieve first x elements from the queue
        next_tasks = redis_client.lrange(TASK_EXEC_LIST_KEY, 0, TASK_CONCURRENCY_LIMIT - 1)
        tasks = []
        if len(next_tasks):
            tasks = [
                queue_task_executor.signature(
                    kwargs={
                        'request_id': int(val),
                        'is_called_by_callback': True
                    }
                )
                for val in next_tasks
            ]
            # delete x elements from queue
            redis_client.ltrim(TASK_EXEC_LIST_KEY, TASK_CONCURRENCY_LIMIT, -1)
            job = group(tasks)
            job.link(queue_task_executor_callback.signature(kwargs={'pick_next': True}))
            _ = job.apply_async()
        else:
            # this means this was the last callback
            redis_client.delete(TASK_EXECUTOR_KEY)
    else:
        logger.debug(f'queue_task_executor_callback already in progress {redis_client.get(TASK_EXECUTOR_KEY)}')

In this function, we retrieve the tasks from the Redis list that were added in the 1st function. The number of tasks to retrieve depends on the concurrency limit we set as TASK_CONCURRENCY_LIMIT. Then, we leverage the Celery Groups feature to execute the tasks in parallel with a callback function that calls this function again with an argument to pick the next bunch of tasks. To avoid duplicate calls, we wrap this whole logic under a lock key TASK_EXECUTOR_KEY, ensuring that only one celery task executes this function queue_task_executor_callback to pick the next bunch of tasks from Redis list.

That's pretty much it! Now, we can add tasks to the queue by calling queue_task_executor celery task.

💡 Advantages of This Redis Queue Setup vs. Cloud Infrastructure

Cost-Effectiveness: Cloud queuing services charge based on usage (e.g., number of requests, message size, or data transfer). With Redis, we pay only for hosting your Redis server, which can be significantly cheaper.
Simplicity: We only need Redis as a dependency. There's no need to set up additional accounts, APIs, or permissions as we would with cloud services, and hence, it's easy to run and debug locally.
Configurable Concurrency: This Redis-based setup allows us to directly control concurrency using the queue_task_executor_callback function.

📢 Final Words

This Redis-based queue system offers a lightweight, flexible, and cost-effective alternative to cloud-based queuing solutions. It empowers developers with full control over task execution, low-latency performance, and the ability to run independently of external vendors.

📌 You can check out my GitHub repository for a complete working example of this approach 👇

idris-rampurawala / redis-queue

Redis Queue project is aimed at creating a queue based in redis and celery using redis data structures.

Redis Queue

Redis Queue project is aimed at creating a queue based in redis and celery using redis data structures. We can control the task concurrency as per our need

Note: This project is forked from one of my project Flask Boilerplate to quickly get started.

Contributing

We encourage you to contribute to this project! Please check out the Contributing guidelines about how to proceed.

Getting Started

Prerequisites

Python 3.11.3 or higher
Up and running Redis client

Project setup

# clone the repo
$ git clone git@github.com:idris-rampurawala/redis-queue.git
# move to the project folder
$ cd redis-queue

If you want to install redis via docker

$ docker run -d --name="redis-queue" -p 6379:6379 redis

Creating virtual environment

Setup the correct python version using pyenv pyenv install 3.11.3
Install pipenv a global python project pip install pipenv
Create a virtual environment for this project
```
# activating the pipenv environment
```
…

View on GitHub

See ya! until my next post 😋

Handling Duplicates in BigQuery

Idris Rampurawala — Wed, 24 May 2023 13:01:37 +0000

BigQuery is a fully managed enterprise data warehouse that helps us manage and analyze our data. It is a prominent product offered by Google Cloud Platform (GCP). BigQuery's scalable, distributed analysis engine lets us query terabytes and petabytes in minutes. Having duplicate rows of data can cause incorrect aggregates or query results. There are various methods to deal with data duplication in BigQuery. We will explore a couple of them in this post.

Consider a scenario where we schedule a job that runs once a day to load data from a data source in BigQuery. That is, each day's data is unique. If for any reason, this job fails in the middle of the processing, then we will have to re-run the job. Now, re-running it could add duplicate rows which would have been inserted before this failure. This duplication of rows may hamper our analytics, reports, etc. that depends on this dataset. How do we handle these duplicate rows? 🤔 Let's take a look.

1. Deleting the duplicates before a re-run

The simplest way out here is to implement the processing in a way that makes sure the current day's data do not exist before inserting. That is, we first delete the current day's data before re-running the job. Now, this may sound like an easy task, but in the world of concurrency, it becomes difficult to scale in the future.

2. Selecting the unique records

Since BigQuery can handle terabytes of data, we can allow data duplication in INSERTs and remove (skip) duplicates in SELECT queries. Oftentimes, various non-technical teams use BigQuery to generate different reports and this duplication may generate unwanted results. Guess what? We have Views to the rescue 😎. Views can abstract the complexity of fetching unique rows on top of the actual dataset.

In general, there are two types of views - Logical and Materialized. See the comparison chart below to select an appropriate option based on your use case.

🔹 Logical Views vs Materialized Views

Feature	Logical Views	Materialized Views
Definition	The query that defines the view is run each time the view is queried.	Views are precomputed views that periodically cache the results of a query for increased performance and efficiency. It stores results in a separate table for caching the results.
Cost	Queries are billed according to the total amount of data in all table fields referenced directly or indirectly by the top-level query.	Cost associated with these views are - Querying materialized view - Maintaining materialized views (refreshing records) - Storing materialized view tables
Storage	No extra storage required.	Requires its own storage.
Maintenance	No maintenance required.	Refreshing the records of materialized views can be either manual or automatic. If manual, then we have to maintain it. Automatic refreshes have cost associated with them.

👉 If you are not generating aggregations, then go for the Logical view. For this post, I will focus on the Logical view that projects the unique rows on top of the dataset.

🔹 Logical View Syntax

CREATE OR REPLACE VIEW
  `table_name_view` OPTIONS ( friendly_name="table_name_view",
    description="A view that fetches latest unique entries processed from table_name" ) AS
SELECT
  *
FROM
  `dataset.table_name`
WHERE
  TRUE QUALIFY ROW_NUMBER() OVER(PARTITION BY column_1, column_2 ORDER BY inserted_at DESC ) = 1;

🔍 In the above query, we are addressing two major issues:

finding unique rows based on multiple key columns using the PARTITION BY
selecting the latest record using inserted_at column that records the row insertion timestamp

💪 Our team has been relying on this View syntax for quite some time now and it has proven to be a valuable asset for our production process.

📑 References

See ya! until my next post 😋

Underrated Skills of a Senior Software Engineer

Idris Rampurawala — Tue, 15 Nov 2022 14:27:05 +0000

Software engineering is a field with various verticals and domains. Each of these domains requires some specific skillset for a software engineer to succeed in their role.

It's reasonable to recognize software engineers' value by their technical skills. But we often undermine that along with technical skills, non-technical skills are equally important. Based on my observation throughout my work experience, I find the following skills which are often underrated.

👁️‍🗨️ Attention To Detail

One of the admirable qualities of a senior role is their ability to foresee the impact in all areas, no matter how small a change/feature. They bring all their expertise to those brainstorming sessions to help you figure out future consequences. This helps create a safe environment for the team to hear and resolve various concerns.

🛠️ Optimize Dev Processes

The senior engineer always makes their presence count with one of the areas you will find them helping most is optimizing dev processes. They always keep track of the difficulties observed in the process and later help to overcome them. They not only identify the problem areas but also proposes options for their resolution to make the life of another engineer hassle-free.

⚖️ Minimize Over-engineering

Sometimes, a software engineer might over-engineer a task/project in their intent to make it more scalable and reliable in the future. An experienced engineer will always question the scope of the project and tries to balance the implementation. This helps the team achieve the results with minimum effort keeping the scope for future enhancements.

🔎 Problem Discovery

A senior engineer in the team always looks out for any problems in the current system. This skill always helps you discover new opportunities within the project. After spending some days or months working on a product, they can better understand the friction points and proposes new ideas/features. This helps build product backlog and keep tech debt on track.

📄 Requirement Focused

An experienced engineer will always focus more on refining the requirements before moving to the implementation. They always make sure to understand what's the ask by clarifying the requirements. This helps the product team to understand any bottlenecks in the initial phases of the scoping.

💬 Conclusion

It's natural to give more importance to the technical skills of a software engineer. But the skills mentioned above are equally dominant to achieve greater heights in their role that also helps the team succeed. I think the time is right to recognize these skills with equal importance!

See ya! Until my next post 😋

Writing Custom Django Database Functions

Idris Rampurawala — Thu, 19 May 2022 11:40:12 +0000

Django's database functions represent functions that will run in the database. It provides a way for users to use functions provided by the underlying database as annotations, aggregations, or filters. Functions are also expressions, so they can be used and combined with other expressions like aggregate functions.

One of the rich Django's features includes customization of its various functionalities. Yes, you got it right! 🙌 We can customize Django database functions as per our needs.

📝 In this post, we will look into a couple of examples below to get the gist of writing custom functions based on our business needs.

👉 Let's first understand the Django Func() class which serves as the basis for us to move forward.

📜 Django `Func(*expressions, **extra)` class

The class Func() is the most general part of Django Query Expressions
It allows the implementation of almost any function or operator into Django ORM in some way
Func() expression is the base type of all expressions that involve database functions like COALESCE and LOWER, or aggregates like SUM
I recommend reading Avoiding SQL injection before using Func()

Following are some ways to write our custom database functions:

🔹 Custom database functions

We can create our custom database functions using Django's Func class. In one of my projects, I wanted to covert the UTC timestamp to IST in Django filter with a specific date format. Writing two simple Django database functions helped me reuse it at multiple instances as follows:

from django.db.models import Func

class TimestampToIST(Func):
    """ Converts the db (UTC) timestamp value to IST equivalent timestamp
    """
    function = 'timezone'
    template = "%(function)s('Asia/Calcutta', %(expressions)s)"


class TimestampToStr(Func):
    """ Converts the timestamp to string using the given format
    """
    function = 'to_char'
    template = "%(function)s(%(expressions)s, 'DD/MM/YYYY HH24:MI:SS')"  # 21/06/2021 16:08:34

# Usage
Author.objects.annotate(last_updated=TimestampToStr(TimestampToIST(F('updated_at'))))

🔹 Partial implementation of database functions

Another great customization example is to make a new version of the function with one or two arguments already filled in. For example, let's create a specialized SubStr that extracts the first character from a string:

from functools import partial
from django.db.models.functions import Substr

ExtractFirstChar = partial(Substr, pos=1, length=1)

# Usage
User.objects.annotate(name_initial=ExtractFirstChar('first_name'))

🔹 Executing a `GROUP BY` without an aggregation function

Imagine a situation where we want to use GROUP BY without using any aggregate function. Django ORM does not allow us to use GROUP BY without an aggregate function 🤨 Hence, to accomplish this, we can create a Django function that is treated as an aggregate function by Django but evaluates to NULL in a SQL query, as sourced from StackOverflow

from django.db.models import CharField, Func

class NullAgg(Func):
    """Annotation that causes GROUP BY without aggregating.

    A fake aggregate Func class that can be used in an annotation to cause
    a query to perform a GROUP BY without also performing an aggregate
    operation that would require the server to enumerate all rows in every
    group.

    Takes no constructor arguments and produces a value of NULL.

    Example:
        ContentType.objects.values('app_label').annotate(na=NullAgg())
    """
    template = 'NULL'
    contains_aggregate = True
    window_compatible = False
    arity = 0
    output_field = CharField()

📑 Resources

See ya! until my next post 😋

Parallelize Processing a Large AWS S3 File

Idris Rampurawala — Fri, 25 Jun 2021 11:15:54 +0000

In my last post, we discussed achieving the efficiency in processing a large AWS S3 file via S3 select. The processing was kind of sequential and it might take ages for a large file. So how do we parallelize the processing across multiple units? 🤔 Well, in this post we gonna implement it and see it working!

📝 I highly recommend checking out my last post on streaming S3 file via S3-Select to set the context for this post.

I always like to break down a problem into the smaller pieces necessary to solve it (analytical approach). Let's try to solve this in 3 simple steps:

1. Find the total bytes of the S3 file

Very similar to the 1st step of our last post, here as well we try to find file size first.
The following code snippet showcases the function that will perform a HEAD request on our S3 file and determines the file size in bytes.



# core/utils.py

def get_s3_file_size(bucket: str, key: str) -> int:
    """Gets the file size of S3 object by a HEAD request

    Args:
        bucket (str): S3 bucket
        key (str): S3 object path

    Returns:
        int: File size in bytes. Defaults to 0 if any error.
    """
    aws_profile = current_app.config.get('AWS_PROFILE_NAME')
    s3_client = boto3.session.Session(profile_name=aws_profile).client('s3')
    file_size = 0
    try:
        response = s3_client.head_object(Bucket=bucket, Key=key)
        if response:
            file_size = int(response.get('ResponseMetadata').get('HTTPHeaders').get('content-length'))
    except ClientError:
        logger.exception(f'Client error reading S3 file {bucket} : {key}')
    return file_size

2. Create a celery task to process a chunk

Here, we would define a celery task to process a file chunk (which will be executed in parallel later). The overall processing here will look like this:

Receive the start and end bytes of this chunk as an argument
Fetch this part of the S3 file via S3-Select and store it locally in a temporary file (as CSV in this example)
Read this temporary file and perform any processing required
Delete this temporary file

📝 I term this task as a file chunk processor. It processes a chunk from a file. Running multiple of these tasks completes the processing of the whole file.



# core/tasks.py

@celery.task(name='core.tasks.chunk_file_processor', bind=True)
def chunk_file_processor(self, **kwargs):
    """ Creates and process a single file chunk based on S3 Select ScanRange start and end bytes
    """
    bucket = kwargs.get('bucket')
    key = kwargs.get('key')
    filename = kwargs.get('filename')
    start_byte_range = kwargs.get('start_byte_range')
    end_byte_range = kwargs.get('end_byte_range')
    header_row_str = kwargs.get('header_row_str')
    local_file = filename.replace('.csv', f'.{start_byte_range}.csv')
    file_path = path.join(current_app.config.get('BASE_DIR'), 'temp', local_file)

    logger.info(f'Processing {filename} chunk range {start_byte_range} -> {end_byte_range}')
    try:
        # 1. fetch data from S3 and store it in a file
        store_scrm_file_s3_content_in_local_file(
            bucket=bucket, key=key, file_path=file_path, start_range=start_byte_range,
            end_range=end_byte_range, delimiter=S3_FILE_DELIMITER, header_row=header_row_str)

        # 2. Process the chunk file in temp folder
        id_set = set()
        with open(file_path) as csv_file:
            csv_reader = csv.DictReader(csv_file, delimiter=S3_FILE_DELIMITER)
            for row in csv_reader:
                # perform any other processing here
                id_set.add(int(row.get('id')))
        logger.info(f'{min(id_set)} --> {max(id_set)}')

        # 3. delete local file
        if path.exists(file_path):
            unlink(file_path)
    except Exception:
        logger.exception(f'Error in file processor: {filename}')

3. Execute multiple celery tasks in parallel

This is the most interesting step in this flow. We will create multiple celery tasks to run in parallel via Celery Group.
Once we know the total bytes of a file in S3 (from step 1), we calculate start and end bytes for the chunk and call the task we created in step 2 via the celery group. The start and end bytes range is a continuous range of file size. Optionally, we can also call a callback (result) task once all our processing tasks get completed.



# core/tasks.py

@celery.task(name='core.tasks.s3_parallel_file_processing', bind=True)
def s3_parallel_file_processing_task(self, **kwargs):
    """ Creates celery tasks to process chunks of file in parallel
    """
    bucket = kwargs.get('bucket')
    key = kwargs.get('key')
    try:
        filename = key
        # 1. Check file headers for validity -> if failed, stop processing
        desired_row_headers = (
            'id',
            'name',
            'age',
            'latitude',
            'longitude',
            'monthly_income',
            'experienced'
        )
        is_headers_valid, header_row_str = validate_scrm_file_headers_via_s3_select(
            bucket=bucket,
            key=key,
            delimiter=S3_FILE_DELIMITER,
            desired_headers=desired_row_headers)
        if not is_headers_valid:
            logger.error(f'{filename} file headers validation failed')
            return False
        logger.info(f'{filename} file headers validation successful')

        # 2. fetch file size via S3 HEAD
        file_size = get_s3_file_size(bucket=bucket, key=key)
        if not file_size:
            logger.error(f'{filename} file size invalid {file_size}')
            return False
        logger.info(f'We are processing {filename} file about {file_size} bytes :-o')

        # 2. Create celery group tasks for chunk of this file size for parallel processing
        start_range = 0
        end_range = min(S3_FILE_PROCESSING_CHUNK_SIZE, file_size)
        tasks = []
        while start_range < file_size:
            tasks.append(
                chunk_file_processor.signature(
                    kwargs={
                        'bucket': bucket,
                        'key': key,
                        'filename': filename,
                        'start_byte_range': start_range,
                        'end_byte_range': end_range,
                        'header_row_str': header_row_str
                    }
                )
            )
            start_range = end_range
            end_range = end_range + min(S3_FILE_PROCESSING_CHUNK_SIZE, file_size - end_range)
        job = (group(tasks) | chunk_file_processor_callback.s(data={'filename': filename}))
        _ = job.apply_async()
    except Exception:
        logger.exception(f'Error processing file: {filename}')


@celery.task(name='core.tasks.chunk_file_processor_callback', bind=True, ignore_result=False)
def chunk_file_processor_callback(self, *args, **kwargs):
    """ Callback task called post chunk_file_processor()
    """
    logger.info('Callback called')



# core/utils.py

def store_scrm_file_s3_content_in_local_file(bucket: str, key: str, file_path: str, start_range: int, end_range: int,
                                             delimiter: str, header_row: str):
    """Retrieves S3 file content via S3 Select ScanRange and store it in a local file.
       Make sure the header validation is done before calling this.

    Args:
        bucket (str): S3 bucket
        key (str): S3 key
        file_path (str): Local file path to store the contents
        start_range (int): Start range of ScanRange parameter of S3 Select
        end_range (int): End range of ScanRange parameter of S3 Select
        delimiter (str): S3 file delimiter
        header_row (str): Header row of the local file. This will be inserted as first line in local file.
    """
    aws_profile = current_app.config.get('AWS_PROFILE_NAME')
    s3_client = boto3.session.Session(profile_name=aws_profile).client('s3')
    expression = 'SELECT * FROM S3Object'
    try:
        response = s3_client.select_object_content(
            Bucket=bucket,
            Key=key,
            ExpressionType='SQL',
            Expression=expression,
            InputSerialization={
                'CSV': {
                    'FileHeaderInfo': 'USE',
                    'FieldDelimiter': delimiter,
                    'RecordDelimiter': '\n'
                }
            },
            OutputSerialization={
                'CSV': {
                    'FieldDelimiter': delimiter,
                    'RecordDelimiter': '\n',
                },
            },
            ScanRange={
                'Start': start_range,
                'End': end_range
            },
        )

        """
        select_object_content() response is an event stream that can be looped to concatenate the overall result set
        """
        f = open(file_path, 'wb')  # we receive data in bytes and hence opening file in bytes
        f.write(header_row.encode())
        f.write('\n'.encode())
        for event in response['Payload']:
            if records := event.get('Records'):
                f.write(records['Payload'])
        f.close()
    except ClientError:
        logger.exception(f'Client error reading S3 file {bucket} : {key}')
    except Exception:
        logger.exception(f'Error reading S3 file {bucket} : {key}')

That's it! 😎 Now, instead of streaming the S3 file bytes by bytes, we parallelize the processing by concurrently processing the chunks. It wasn't that tough, isn't it? 😅

🔍 Comparing the processing time

If we compare the processing time of the same file we processed in our last post with this approach, the processing runs approximately 68% faster (with the same hardware and config). 😆

	Streaming S3 File	Parallel Processing S3 File
File size	4.8MB	4.8MB
Processing time	~37 seconds	~12 seconds

✔️ Benefits of this approach

A very large file containing millions of records can be processed within minutes. I have been using this approach in the production environment for a while, and it's very blissful
Computing and processing is distributed among distributed workers
Processing speed can be tweaked by the availability of worker pools
No more memory issues

📌 You can check out my GitHub repository for a complete working example of this approach 👇

idris-rampurawala / s3-select-demo

This project showcases the rich AWS S3 Select feature to stream a large data file in a paginated style.

AWS S3 Select Demo

This project showcases the rich AWS S3 Select feature to stream a large data file in a paginated style.

Currently, S3 Select does not support OFFSET and hence we cannot paginate the results of the query. Hence, we use scanrange feature to stream the contents of the S3 file.

Background

Importing (reading) a large file leads Out of Memory error. It can also lead to a system crash event. There are libraries viz. Pandas, Dask, etc. which are very good at processing large files but again the file is to be present locally i.e. we will have to import it from S3 to our local machine. But what if we do not want to fetch and store the whole S3 file locally at once? 🤔

Well, we can make use of AWS S3 Select to stream a large file via it's ScanRange parameter. This approach…

View on GitHub

📑 Resources

See ya! until my next post 😋

Efficiently Streaming a Large AWS S3 File via S3 Select

Idris Rampurawala — Tue, 06 Apr 2021 13:37:57 +0000

AWS S3 is an industry-leading object storage service. We tend to store lots of data files on S3 and at times require processing these files. If the size of the file that we are processing is small, we can basically go with traditional file processing flow, wherein we fetch the file from S3 and then process it row by row level. But the question arises, what if the file is size is more viz. > 1GB? 😓

📜 Let's consider some of the use-cases:

We want to process a large CSV S3 file (~2GB) every day. It must be processed within a certain time frame (e.g. in 4 hours)
We are required to process large S3 files regularly from the FTP server. New files come in certain time intervals and to be processed sequentially i.e. the old file has to be processed before starting to process the newer files.

These are some very good scenarios where local processing may impact the overall flow of the system. Also, if we are running these file processing units in containers, then we have got limited disk space to work with. Hence, a cloud streaming flow is needed (which can also parallelize the processing of multiple chunks of the same file by streaming different chunks of the same file in parallel threads/processes). This is where I came across the AWS S3 Select feature. 😎

📝 This post focuses on streaming a large file into smaller manageable chunks (sequentially). This approach can then be used to parallelize the processing by running in concurrent threads/processes. Check my next post on this.

S3 Select

With Amazon S3 Select, you can use simple structured query language (SQL) statements to filter the contents of Amazon S3 objects and retrieve just the subset of data that you need. Using Amazon S3 Select to filter this data, you can reduce the amount of data that Amazon S3 transfers, reducing the cost and latency to retrieve this data.

Amazon S3 Select works on objects stored in CSV, JSON, or Apache Parquet format. It also works with objects that are compressed with GZIP or BZIP2 (for CSV and JSON objects only) and server-side encrypted objects. You can specify the format of the results as either CSV or JSON, and you can determine how the records in the result are delimited.

📝 We will be using Python's boto3 to accomplish our end goal.

🧱 Constructing SQL expressions

In order to work with S3 Select, boto3 provides select_object_content() function to query S3. You pass SQL expressions to Amazon S3 in the request. Amazon S3 Select supports a subset of SQL. Check this link for more information on this.



response = s3_client.select_object_content(
    Bucket=bucket,
    Key=key,
    ExpressionType='SQL',
    Expression='SELECT * FROM S3Object',
    InputSerialization={
        'CSV': {
            'FileHeaderInfo': 'USE',
            'FieldDelimiter': ',',
            'RecordDelimiter': '\n'
        }
    },
    OutputSerialization={
        'JSON': {
            'RecordDelimiter': ','
        }
    }
)

In above request, InputSerialization determines the S3 file type and related properties, while OutputSerialization determines the response that we get out of this select_object_content().

🌫️ Streaming Chunks

Now, as we have got some idea about how the S3 Select works, let's try to accomplish our use-case of streaming chunks (subset) of a large file like how a paginated API works. 😋

S3 Select supports ScanRange parameter which helps us to stream a subset of an object by specifying a range of bytes to query. S3 Select requests for a series of non-overlapping scan ranges. Scan ranges don't need to be aligned with record boundaries. A record that starts within the scan range specified but extends beyond the scan range will be processed by the query. It means that the row would be fetched within the scan range and it might extend to fetch the whole row. It doesn't fetch a subset of a row, either the whole row is fetched or it is skipped (to be fetched in another scan range).

Let's try to achieve this in 2 simple steps:

1. Find the total bytes of the S3 file

The following code snippet showcases the function that will perform a HEAD request on our S3 file and determines the file size in bytes.



def get_s3_file_size(bucket: str, key: str) -> int:
    """Gets the file size of S3 object by a HEAD request

    Args:
        bucket (str): S3 bucket
        key (str): S3 object path

    Returns:
        int: File size in bytes. Defaults to 0 if any error.
    """
    aws_profile = current_app.config.get('AWS_PROFILE_NAME')
    s3_client = boto3.session.Session(profile_name=aws_profile).client('s3')
    file_size = 0
    try:
        response = s3_client.head_object(Bucket=bucket, Key=key)
        if response:
            file_size = int(response.get('ResponseMetadata').get('HTTPHeaders').get('content-length'))
    except ClientError:
        logger.exception(f'Client error reading S3 file {bucket} : {key}')
    return file_size

2. Create a generator to stream the chunks

Now, the logic is to yield the chunks of byte stream of the S3 file until we reach the file size. Rest assured, this continuous scan range won't result in over-lapping of rows in the response 😉 (check the output image / GitHub repo). Simple enough, eh? 😝



import ast
import boto3
from botocore.exceptions import ClientError

def stream_s3_file(bucket: str, key: str, file_size: int, chunk_bytes=5000) -> tuple[dict]:
    """Streams a S3 file via a generator.

    Args:
        bucket (str): S3 bucket
        key (str): S3 object path
        chunk_bytes (int): Chunk size in bytes. Defaults to 5000
    Returns:
        tuple[dict]: Returns a tuple of dictionary containing rows of file content
    """
    aws_profile = current_app.config.get('AWS_PROFILE_NAME')
    s3_client = boto3.session.Session(profile_name=aws_profile).client('s3')
    expression = 'SELECT * FROM S3Object'
    start_range = 0
    end_range = min(chunk_bytes, file_size)
    while start_range < file_size:
        response = s3_client.select_object_content(
            Bucket=bucket,
            Key=key,
            ExpressionType='SQL',
            Expression=expression,
            InputSerialization={
                'CSV': {
                    'FileHeaderInfo': 'USE',
                    'FieldDelimiter': ',',
                    'RecordDelimiter': '\n'
                }
            },
            OutputSerialization={
                'JSON': {
                    'RecordDelimiter': ','
                }
            },
            ScanRange={
                'Start': start_range,
                'End': end_range
            },
        )

        """
        select_object_content() response is an event stream that can be looped to concatenate the overall result set
        Hence, we are joining the results of the stream in a string before converting it to a tuple of dict
        """
        result_stream = []
        for event in response['Payload']:
            if records := event.get('Records'):
                result_stream.append(records['Payload'].decode('utf-8'))
        yield ast.literal_eval(''.join(result_stream))
        start_range = end_range
        end_range = end_range + min(chunk_bytes, file_size - end_range)


def s3_file_processing():
    bucket = '<s3-bucket>'
    key = '<s3-key>'
    file_size = get_s3_file_size(bucket=bucket, key=key)
    logger.debug(f'Initiating streaming file of {file_size} bytes')
    chunk_size = 524288  # 512KB or 0.5MB
    for file_chunk in stream_s3_file(bucket=bucket, key=key,
                                     file_size=file_size, chunk_bytes=chunk_size):
        logger.info(f'\n{30 * "*"} New chunk {30 * "*"}')
        id_set = set()
        for row in file_chunk:
            # perform any other processing here
            id_set.add(int(row.get('id')))
        logger.info(f'{min(id_set)} --> {max(id_set)}')

Congratulations! 👏 We have successfully managed to solve one of the key challenges of processing a large S3 file without crashing our system. 🤘

📌 You can check out my GitHub repository for a complete working example of this approach 👇

idris-rampurawala / s3-select-demo

This project showcases the rich AWS S3 Select feature to stream a large data file in a paginated style.

AWS S3 Select Demo

This project showcases the rich AWS S3 Select feature to stream a large data file in a paginated style.

Currently, S3 Select does not support OFFSET and hence we cannot paginate the results of the query. Hence, we use scanrange feature to stream the contents of the S3 file.

Background

Well, we can make use of AWS S3 Select to stream a large file via it's ScanRange parameter. This approach…

View on GitHub

✔️ Benefits of using S3-Select

Reduced IO — thus better performance
Reduced costs due to smaller data transfer fees
Multiple chunks can be run in parallel to expedite the file processing using ScanRange in multiple threads/processes

❗ Limitations of S3 Select

The maximum length of a record in the input or result is 1 MB
Amazon S3 Select can only emit nested data using the JSON output format
S3 select returns a stream of encoded bytes, so we have to loop over the returned stream and decode the output records['Payload'].decode('utf-8')
Only works on objects stored in CSV, JSON, or Apache Parquet format. For more flexibility/features, you can go for AWS Athena

📑 Resources

You might also wanna read a sequel of this post 👇

Parallelize Processing a Large AWS S3 File

Idris Rampurawala ・ Jun 25 '21

#aws #python #showdev #productivity

See ya! until my next post 😋

User Flow with dropouts using D3 Sankey in Angular 10

Idris Rampurawala — Thu, 17 Dec 2020 13:00:20 +0000

📜 Background

A Sankey chart (diagram) is a visualization used to depict a flow from one set of values to another. The things being connected are called nodes and the connections are called links.

One of the use cases of the Sankey chart is to represent the user flow or user journey or screen flow. When it comes to user journey analysis, these diagrams allow identifying at a glance what are the most frequently fired events, in what order, or what are the different paths from action A to action B. These are information that marketers, decision-makers, or your customers are likely to be interested in. Specifically, when representing user journeys, the dropouts are a game-changer.

A dropout is basically a node representing the number of users that didn't go to the next screen/action but exited the application (user flow). There are a lot of open-source libraries available to create Sankey charts such as Google, Highcharts, D3, etc. But none of them provide a capability to put in a dropout node in the flow. I was able to accomplish this using D3's Sankey chart. Let's check out how I achieved this feat 😁

✅ Prerequisites

We will assume that you have a basic knowledge of the Angular framework and D3 Sankey chart
This post only aims to guide with the logic of implementation and hence showcasing only code snippets. For overall code implementation, check out my GitHub repository.

🧱 Integrating D3 with Angular

1. Install D3

D3 is hosted on npm so we can easily install it via the npm command
npm install d3

2. Install D3-sankey

To create Sankey charts in D3, we will have to add d3-sankey npm package on top of the D3
npm install d3-sankey

3. Import dependencies in Angular

In our app.component.ts, we will just import packages as

import * as d3 from 'd3';
import * as d3Sankey from 'd3-sankey';

That's it! We are now ready to move to the next step 😁

⚒️ Implementing Dropouts in Sankey chart

A Sankey chart consists of two entities to generate the graph:

Node	A rectangular box that represents the actual entity (i.e. in our example, it represents a screen user is visiting)
Link	It connects two nodes based on their weight

1. Preparing data

The minimum attributes that are required to create a node are node (unique id), name. Here, we will also add one more attribute drop ➖ a number representing the dropouts on this node.
Similarly, for links, the attributes are source, target, value.
To represent dropouts, we will make a node with name as Dropout and drop as 0. This node will not have any link that will result in two nodes (node and dropout node) placed adjacent to each other.

Our data structure would look like this:

'nodes': [
      {
        'node': 0, // unique node id
        'name': 'HomeActivity', // name of the node
        'drop': 2 // weight representing if any dropouts from this node
      },
      {
        'node': 1,
        'name': 'Dropout', // we always represent dropouts with this common name
        'drop': 0
      },
...
]
'links': [
      {
        'source': 0, // source node id
        'target': 1, // target node id
        'value': 2 // link weight
      },
      {
        'source': 0,
        'target': 2,
        'value': 2
      },
...
]

2. Prepare HTML for rendering

Once we have data generated, it's time to add Sankey chart logic to generate the graph.

Let's consider we have a div for plotting sankey

<!-- app.component.html -->
 <div id="sankey"></div>

3. Adding Sankey chart rendering logic

Secondly, let's add some initial Sankey chart rendering logic in the app.component.ts ngOnInit function which gets called on-page init

// app.component.ts
...

ngOnInit(): void {
  // get some dummy data created in above step
  const chartData = {
    'nodes': [
      {
        'node': 0, // unique node id
        'name': 'HomeActivity', // name of the node
        'drop': 2 
      },
      ...
    ],
    'links': [
      {
        'source': 0,
        'target': 1,
        'value': 2
      }
      ...
    ]
  };

  this.drawChart(chartData);
}
...

drawChart(chartData): void {
 // plotting the sankey chart
    const sankey = d3Sankey.sankey()
      .nodeWidth(15)
      .nodePadding(10)
      .nodeAlign(d3Sankey.sankeyLeft)
      .extent([[1, 1], [width, height]]);
    sankey(chartData);

    const iter = d3.nest()
      .key((d: any) => d.x0)
      .sortKeys(d3.ascending)
      .entries(chartData.nodes)
      .map((d: any) => d.key)
      .sort((a: any, b: any) => a - b);

    // add svg for graph
    const svg = d3.select('#sankey').append('svg')
      .attr('width', width)
      .attr('height', height)
      .attr('viewbox', `0 0 ${width} ${height}`);
}

4. Adding Sankey chart links

Now, let's add links to the Sankey chart (in the same drawChart()). We are gonna exclude the links which end with the Dropout node i.e. the links that have target as the Dropout node. This will help us to create a dropout node adjacent to its source node without any link in between.

// app.component.ts
drawChart(chartData): void {
...
// add in the links (excluding the dropouts, coz it will become node)
    const link = svg.append('g')
      .selectAll('.link')
      .data(chartData.links)
      .enter()
      .filter((l: any) => l.target.name.toLowerCase() !== DROPOUT_NODE_NAME)
      .append('path')
      .attr('d', d3Sankey.sankeyLinkHorizontal()
      )
      .attr('fill', 'none')
      .attr('stroke', '#9e9e9e')
      .style('opacity', '0.7')
      .attr('stroke-width', (d: any) => Math.max(1, d.width))
      .attr('class', 'link')
      .sort((a: any, b: any) => {
        if (a.target.name.toLowerCase() === DROPOUT_NODE_NAME) {
          return -1;
        } else if (b.target.name.toLowerCase() === DROPOUT_NODE_NAME) {
          return 1;
        } else {
          return 0;
        }
      })
      ;

}

5. Adding dropout nodes

Let's now plot the dropout nodes. This is the most important step as we plot the dropout nodes here. So how do we achieve this? Well, remember, we left the link which targets the dropout node in the above step? That's where we put in the dropout node (i.e. a rectangle in terms of D3).

The most important question is how to identify the height of this dropout node? 😦 It's a little tricky question to solve. Remember, we are plotting dropout at the source node and hence we find the height of all the links on this node, excluding the dropout link (which we have not plotted). So, the dropout node height is
🧐 height of source node - the height of all non-dropout links of this node

// app.component.ts
drawChart(chartData): void {
...

    // plotting dropout nodes
    const dropLink = svg.append('g')
      .selectAll('.link')
      .data(chartData.links)
      .enter()
      .filter((l: any) => l.target.name.toLowerCase() === DROPOUT_NODE_NAME)
      .append('rect')
      .attr('x', (d: any) => d.source.x1)
      .attr('y', (d: any) => {
        if (d.source.drop > 0) {
          let totalWidth = 0;
          for (const elm of d.source.sourceLinks) {
            if (elm.target.name.toLowerCase() === DROPOUT_NODE_NAME) {
              break;
            } else if (elm.value >= d.source.drop && elm.target.name.toLowerCase() !== DROPOUT_NODE_NAME) {
              totalWidth += elm.width;
            }
          }
          return d.source.y0 + totalWidth;
        } else {
          return d.source.y0;
        }
      })
      .attr('height', (d: any) => Math.abs(d.target.y0 - d.target.y1))
      .attr('width', (d: any) => sankey.nodeWidth() + 3)
      .attr('fill', '#f44336')
      .attr('stroke', '#f44336')
      .attr('class', 'dropout-node')
      .on('click', (l: any) => {
        fnOnDropOutLinkClicked(l);
      });

    dropLink.append('title')
      .text((d: any) => d.source.name + '\n' +
        'Dropouts ' + format(d.value));

    // add the link titles
    link.append('title')
      .text((d: any) => d.source.name + ' → ' +
        d.target.name + '\n' + format(d.value));

}

6. Finishing the chart by adding non-dropout nodes

Finally, let's add all the non-dropout nodes

// app.component.ts
drawChart(chartData): void {
...
    // plotting the nodes
    const node = svg.append('g').selectAll('.node')
      .data(chartData.nodes)
      .enter().append('g')
      .attr('class', 'node')
      .on('mouseover', fade(1))
      .on('mouseout', fade(0.7))
      .on('click', (d) => {
        fnOnNodeClicked(d);
      });

    node.append('rect')
      .filter((d: any) => d.name.toLowerCase() !== DROPOUT_NODE_NAME)
      .attr('x', (d: any) => d.x0)
      .attr('y', (d: any) => d.y0)
      .attr('height', (d: any) => d.y1 - d.y0)
      .attr('width', (d: any) => d.x1 - d.x0)
      .attr('fill', '#2196f3')
      .append('title')
      .text((d: any) => d.name + '\n' + format(d.value));

    node.append('text')
      .filter((d: any) => d.name.toLowerCase() !== DROPOUT_NODE_NAME)
      .attr('x', (d: any) => d.x1 + 20)
      .attr('y', (d: any) => (d.y1 + d.y0) / 2)
      .attr('dy', '0.35em')
      .attr('font-size', 10)
      .attr('font-family', 'Roboto')
      .attr('text-anchor', 'end')
      .text((d: any) => truncateText(d.name, 20))
      .attr('text-anchor', 'start')
      .append('title')
      .text((d: any) => d.name);

}

🏆 Voila! That's all we need to create a dropout node feature in Sankey chart 😌

✔️ For more features such as, showing interaction levels, node click handler, dynamic data update, etc you can check my GitHub repository or visit this for a live demo.

If you like my post do not forget to hit ❤️ or 🦄 See ya! until my next post 😋

Exporting Data To Excel and CSV in Angular

Idris Rampurawala — Mon, 31 Aug 2020 13:41:42 +0000

While working on a web application, there are various instances where we would allow our users to download the data into their specific formats. One such requirement is to allow them to export the data in a spreadsheet (excel) or a CSV file.

This is a very common use case and hence I thought to create a step by step guide to accomplish this easily. We would be discussing the export into 2 main areas:

Export to Excel
Export to CSV

📝 NOTE
I have created a repository on GitHub on this implementation

idris-rampurawala / ng-data-export

Demonstration of an export service that exports data to excel, csv in Angular 10

Export to Excel

The ability to export the data to excel not only gives a powerful feature for the users but also the ability to create an array of other related features to help our users better understand the data. So how do we start? Well, as you'd expect, we have an npm package to deal with it - xlsx (also known as sheetjs) 😁

Installing dependencies



# installing xlsx package
$ npm install xlsx
# installing file-saver - a solution to saving files on the client-side
$ npm install file-saver

Creating an export service

One way of creating common functionalities in Angular is by creating a service for it. Hence, we create an export service which will have functions to export all types of information (excel and CSV for this post).

Using xlsx

xlsx provides a very rich set of utilities to create or parse spreadsheets. For the sake of simplicity, we will concentrate on a couple of the utilities here.

1️⃣ Exporting an HTML table

If we want to export an HTML table to excel, then it is pretty easy as xlsx provides utility for it. Consider if we have a table 👇



<!-- app.component.html -->
<table class="table table-sm" #userTable> <!-- we will make use of this angular var as element reference -->
  <thead class="thead-dark">
    <tr>
      <th scope="col">#</th>
      ...
    </tr>
  </thead>
  <tbody>
    <tr *ngFor="let user of users">
      <td scope="row">{{ user.id }}</td>
      ...
    </tr>
    <tr>
  </tbody>
</table>

Now, we can just create a service function to take this HTML element reference and generate the excel from it (using <thead> and <tbody>).



/* export.service.ts */
import { Injectable, ElementRef } from '@angular/core';
import * as FileSaver from 'file-saver';
import * as XLSX from 'xlsx';

const EXCEL_EXTENSION = '.xlsx';

@Injectable()
export class ExportService {
  constructor() { }

  /**
   * Creates excel from the table element reference.
   *
   * @param element DOM table element reference.
   * @param fileName filename to save as.
   */
  public exportTableElmToExcel(element: ElementRef, fileName: string): void {
    const ws: XLSX.WorkSheet = XLSX.utils.table_to_sheet(element.nativeElement);
    // generate workbook and add the worksheet
    const workbook: XLSX.WorkBook = XLSX.utils.book_new();
    XLSX.utils.book_append_sheet(workbook, ws, 'Sheet1');
    // save to file
    XLSX.writeFile(workbook, `${fileName}${EXCEL_EXTENSION}`);

  }
  ...
}

And in component.ts we just create a handler for the export button to make an attempt to save the file as excel on the client machine.



/* app.component.ts */
import { Component, OnInit, ViewChild, ElementRef } from '@angular/core';
import { ExcelJson } from './interfaces/excel-json.interface';
import { ExportService } from './services/export.service';
...

@Component({
  selector: 'app-root',
  templateUrl: './app.component.html',
  styleUrls: ['./app.component.scss']
})
export class AppComponent implements OnInit {
  ...
  /* the table reference */
  @ViewChild('userTable') userTable: ElementRef;
  ...

  constructor(
    private exportService: ExportService
  ) { }

  ngOnInit(): void {
    ...
  }

  /**
   * Function prepares data to pass to export service to create excel from Table DOM reference
   *
   */
  exportElmToExcel(): void {
    this.exportService.exportTableElmToExcel(this.userTable, 'user_data');
  }

  ...

}

That was pretty easy, isn't it? 😆 What if we want to export more complex data? 🙄 Let's find out 👇

2️⃣ Exporting more complex data

xlsx provides various other utilities to customize the data in excel (using excel column name identifies A, B, C..). For example, I had created a function to export the whole dashboard data to excel in one of my projects. Let's create a function in service for the same.



/* export.service.ts */
...

  /**
   * Creates XLSX option from the Json data. Use this to customize the sheet by adding arbitrary rows and columns.
   *
   * @param json Json data to create xlsx.
   * @param fileName filename to save as.
   */
  public exportJsonToExcel(json: ExcelJson[], fileName: string): void {
    // inserting first blank row
    const worksheet: XLSX.WorkSheet = XLSX.utils.json_to_sheet(
      json[0].data,
      this.getOptions(json[0])
    );

    for (let i = 1, length = json.length; i < length; i++) {
      // adding a dummy row for separation
      XLSX.utils.sheet_add_json(
        worksheet,
        [{}],
        this.getOptions(
          {
            data: [],
            skipHeader: true
          }, -1)
      );
      XLSX.utils.sheet_add_json(
        worksheet,
        json[i].data,
        this.getOptions(json[i], -1)
      );
    }
    const workbook: XLSX.WorkBook = { Sheets: { Sheet1: worksheet }, SheetNames: ['Sheet1'] };
    // save to file
    XLSX.writeFile(workbook, `${fileName}${EXCEL_EXTENSION}`);
  }

  /**
   * Creates the XLSX option from the data.
   *
   * @param json Json data to create xlsx.
   * @param origin XLSX option origin.
   * @returns options XLSX options.
   */
  private getOptions(json: ExcelJson, origin?: number): any {
    // adding actual data
    const options = {
      skipHeader: true,
      origin: -1,
      header: []
    };
    options.skipHeader = json.skipHeader ? json.skipHeader : false;
    if (!options.skipHeader && json.header && json.header.length) {
      options.header = json.header;
    }
    if (origin) {
      options.origin = origin ? origin : -1;
    }
    return options;
  }

...

And in component.ts, we create the data in xlsx required format to pass to this service function



/* app.component.ts */
...

/**
   * Function prepares data to pass to export service to create excel from Json
   *
   */
  exportToExcel(): void {

    const edata: Array<ExcelJson> = [];
    const udt: ExcelJson = {
      data: [
        { A: 'User Data' }, // title
        { A: '#', B: 'First Name', C: 'Last Name', D: 'Handle' }, // table header
      ],
      skipHeader: true
    };
    this.users.forEach(user => {
      udt.data.push({
        A: user.id,
        B: user.firstName,
        C: user.lastName,
        D: user.handle
      });
    });
    edata.push(udt);

    // adding more data just to show "how we can keep on adding more data"
    const bd = {
      data: [
        // chart title
        { A: 'Some more data', B: '' },
        { A: '#', B: 'First Name', C: 'Last Name', D: 'Handle' }, // table header
      ],
      skipHeader: true
    };
    this.users.forEach(user => {
      bd.data.push({
        A: String(user.id),
        B: user.firstName,
        C: user.lastName,
        D: user.handle
      });
    });
    edata.push(bd);
    this.exportService.exportJsonToExcel(edata, 'user_data_customized');
  }

...

Explanation

Confused? 😕 Let me explain what we just did there.

xlsx (or spreadsheets) have a workbook (that's an actual file) and in that, we can have multiple sheets added.
xlsx provides a utility function sheet_add_json() to convert an array of objects to excel data with additional xlsx options. Hence, we just created a wrapper around it our service by which we can pass multiple objects with different xlsx options. This way our export service handles the complexity and we are only required to create an array of objects to pass to it.
xlsx expects the array of objects in the form of {cell: value } and hence {A: 'value'} means that we want to put this value in the cell (column) A of the excel.
skipHeader is to skip the auto-generated header from the objects being passed to the function sheet_add_json()
origin: -1 is to append data to the bottom of worksheet starting on the first column
Also, ExcelJson is a custom interface (that I have created) to define the type of data that service function expects. It represents a valid object data for xlsx.

For more information, please read the xlsx documentation and a sample implementation on github.

How do I style excel? 🧐

xlsx does not provide styling in its open-source version. You can opt for a pro version for styling and dedicated support.

Alternatively, xlsx-style is a fork of xlsx that provides styling on top of it.

One more a very popular alternative to xlsx is ExcelJS. It has got styling included as well but provides lesser utilities in comparison to xlsx.

Export to CSV

Now let's move on to the second part of export i.e. CSV.

Don't worry 😟 it's pretty easy. We just need to add a function to our export service which accepts an array of objects along with a column header to create a CSV for it.



/* export.service.ts */
...
 /**
   * Saves the file on the client's machine via FileSaver library.
   *
   * @param buffer The data that need to be saved.
   * @param fileName File name to save as.
   * @param fileType File type to save as.
   */
  private saveAsFile(buffer: any, fileName: string, fileType: string): void {
    const data: Blob = new Blob([buffer], { type: fileType });
    FileSaver.saveAs(data, fileName);
  }

  /**
   * Creates an array of data to CSV. It will automatically generate a title row based on object keys.
   *
   * @param rows array of data to be converted to CSV.
   * @param fileName filename to save as.
   * @param columns array of object properties to convert to CSV. If skipped, then all object properties will be used for CSV.
   */
  public exportToCsv(rows: object[], fileName: string, columns?: string[]): string {
    if (!rows || !rows.length) {
      return;
    }
    const separator = ',';
    const keys = Object.keys(rows[0]).filter(k => {
      if (columns?.length) {
        return columns.includes(k);
      } else {
        return true;
      }
    });
    const csvContent =
      keys.join(separator) +
      '\n' +
      rows.map(row => {
        return keys.map(k => {
          let cell = row[k] === null || row[k] === undefined ? '' : row[k];
          cell = cell instanceof Date
            ? cell.toLocaleString()
            : cell.toString().replace(/"/g, '""');
          if (cell.search(/("|,|\n)/g) >= 0) {
            cell = `"${cell}"`;
          }
          return cell;
        }).join(separator);
      }).join('\n');
    this.saveAsFile(csvContent, `${fileName}${CSV_EXTENSION}`, CSV_TYPE);
  }

...

The code is pretty much self-explanatory 🤓 where we check if any of the column's data present in the data passed, and generates a CSV from it. We can always change the delimiter from , to any other based on our requirement. file-saver package is required to save the file on the client's machine.

Well, that was pretty simple, isn't it? 🙌 You check out my GitHub repo for a complete implementation of this post.

If you find this helpful or have any suggestions, feel free to comment. Also, do not forget to hit ❤️ or 🦄 if you like my post.

See ya! until my next post 😋

Flask Boilerplate - Structuring Flask App

Idris Rampurawala — Thu, 06 Aug 2020 12:04:17 +0000

Flask is a lightweight WSGI web application framework. It is part of the categories of the micro-framework with little to no dependencies on external libraries. It doesn't enforce any dependencies or project layout. Because of which you will find plenty of different project layouts out there.

While I started working on Flask, I got confused to choose the correct layout for my applications. In this post, we will discuss some of the important points to keep in mind while structuring a Flask application with the help of my Flask Boilerplate project.

idris-rampurawala / flask-boilerplate

A Flask Boilerplate to quickly get started with production-grade flask application. It has some additional packages pre-configured for ease of development.

📝 NOTE

This boilerplate was created based on my cumulative knowledge gained by writing multiple flask projects
This may not be the choice of layout for your app as you may have different requirements but a great way to kick start a Flask app and add packages as per the need
It is lightweight in that it only has minimal packages to achieve the necessities of an app. But you can surely customize it based on your need 😉

Why Boilerplate? 🤔

When I started working on a flask application for the first time, I had a tough time figuring out the correct structure and packages to achieve the necessities of my flask app. Hence, I decided to make this very basic lightweight boilerplate available for the community while I figured out the structure working on multiple live projects in the past.

When we start a project from scratch, especially the backend services, there are some basic functionalities that it should possess. The following are some of the points in building up an efficient app explained concerning this flask boilerplate.

Shortcuts
Folder Structure
Managing External Packages
Workspace Setup
Configuration
Logging
Authentication
Database Setup
Task Queues (Celery) Setup
Deployment
API Caching and Validation

1. Folder Structure

Flask applications generally have many variations in their folder structure. The choice of keeping files in different folders is dependent on an individual/team/project type. On a high level, flask application can be created even with a single page or via a factory pattern method. Let's talk a bit more about the concepts here:

Factory Pattern
Well-structured web apps separate logic between files and modules, typically with separation of concerns in mind. This seems tricky with Flask at first glance because our app depends on an app object that we create via app = Flask(__name__). Separating logic between modules means we would be importing this app object all over the place, eventually leading to problems like circular imports. The Flask Application Factory refers to a common pattern for solving this dilemma. The reason why the Application Factory is so important has to do with something called Flask's Application Context. Hence, we are making use of factory pattern inside this boilerplate project to make our lives a bit easier.
Blueprints
In a normal project, the number of views, and templates, and models, and forms, etc. will grow. You can (hackily) separate them, but wouldn’t it be nice to have something that groups related views, templates, models, and forms? Blueprints! They are a great way to modularly compose Flask applications, which will help scale up projects.

Well, in this boilerplate, there's one blueprint core serving as a base and more can be added later as shown below:

# app/__init__.py
def create_app():
    ...
    app = Flask(APP_NAME)
    ...
    from .core.views import core as core_blueprint
    app.register_blueprint(
        core_blueprint,
        url_prefix='/api/v1/core'
    )

So, the folder structure looks as 👇

├── app
│   ├── core                # blueprint
│   │   ├── __init__.py
│   │   ├── constants.py
│   │   ├── enums.py
│   │   ├── tasks.py        # celery tasks specific to this blueprint
│   │   └── views.py
│   ├── config.py
│   └── __init__.py         # app initialization logic
├── .env
├── .gitignore
├── Pipfile
├── Pipfile.lock
├── README.md
├── authentication.py
├── celery_worker.py
└── run.py

2. Managing External Packages

One of the most important steps is to create a virtual environment for python projects to manage dependencies. There are various python packages for this purpose such as virtualenv, pyenv, etc. This boilerplate uses pipenv to address all the issues that I have faced using other packages for managing dependencies. Pipenv automatically creates and manages a virtual environment for your projects, as well as adds or removes packages from your Pipfile as you install/uninstall packages.

This boilerplate has those packages which are required to build up an app and hence you can always customize based on your need 😉

3. Workspace Setup

Though not directly related to an app, your workspace setup does affect your app creation process. Hence, setting up a workspace is a major step to speed up your working environment. I have written a post that specifically talks about setting up a python workspace in a visual studio code editor which you can refer and quickly get started ✌️

Setting up Python workspace in Visual Studio Code (vscode)

Idris Rampurawala ・ Mar 10 '20

#beginners #python #vscode

4. Configuration

Configuring a Flask app (or maybe any app) generally requires a lot of effort (if we are not experienced). In general, there can be 2 aspects to it -

Storing secrets securely
Usually, a lot of secrets makes up a project such as DB credentials, API keys, etc. It is recommended to keep these secrets in a secured location not being tracked in git. Which is why we have a .env (git untracked, to be created from .env.example) file at the root location of this boilerplate which makes all the secrets present in it available in our project using python-dotenv package. Hence, it is super easy to get started with just creating .env from .env.example by setting correct values 😁
Creating multiple configurations (dev, staging, prod)
Another important point is to separate global configurations based on the environment such as dev, staging, prod. That's the reason we have config.py, a file containing configuration classes based on the environments that get loaded in the Flask app based on the current working environment (which can be set in .env).

# configuration classes (app/config.py)
class BaseConfig(object):
    ''' Base config class. '''
...

class Development(BaseConfig):
    ''' Development config. '''
...

class Staging(BaseConfig):
    ''' Staging config. '''
...

class Production(BaseConfig):
    ''' Production config '''
...
config = {
    'development': Development,
    'staging': Staging,
    'production': Production,
}

# app/__init__.py
from config import config
...

def create_app():
    # loading env vars from .env file
    load_dotenv()
    APPLICATION_ENV = get_environment()
    app = Flask(APP_NAME)
    # loading config based on current environment
    app.config.from_object(config[APPLICATION_ENV])
    # APPLICATION_ENV is set in .env to set current environment

5. Logging

Logging is a means of tracking events that happen when some software runs. We add logging calls to our code to indicate that certain events have occurred.

As we already have environment based configurations in our app, all the other configurations can be changed according to the current app environment (dev, staging, prod). The loggers are configured in config.py and can be overridden in the respective app environments (dev, staging, prod) as shown below 👇

# config.py
class BaseConfig(object):
    ''' Base config class. '''
    ...
    LOG_INFO_FILE = path.join(basedir, 'log', 'info.log')
    LOG_CELERY_FILE = path.join(basedir, 'log', 'celery.log')
    LOGGING = {
        'version': 1,
    ...

6. Authentication

Authentication can be added authentication as per the need of the application. To start with, this boilerplate has API-Key based authentication whereas the API-Key is declared in .env file. In file, authentication.py, a decorator looks for x-api-key header in the request and returns 401 in case of mismatch. Once you have set up the app, you can test the test route (/core/restricted) for the same via curl with your API key set in .env.

$ curl --location --request GET 'http://localhost:5000/api/v1/core/restricted' --header 'x-api-key: <your-api-key>'

For more complex authentication logic, please write a custom middleware.

7. Database Setup

To keep the boilerplate as light as possible of the packages, it does not have any database package added. But worry not if you want to add one. 😌 It is pretty easy to configure the database as this boilerplate is created via the factory pattern method along with environment-based configuration as discussed in the above points. Let's take an example of one of the popular library - SQLAlchemy.

# app/__init__.py

from flask_sqlalchemy import SQLAlchemy

db = SQLAlchemy()

def create_app(config_filename):
    ...
    app = Flask(__name__)
    app.config.from_object(config[APPLICATION_ENV])
    ...
    db.init_app(app)

The configuration would be read directly from app.config which we have loaded from config.py.

8. Task Queues (Celery) Setup

Task queues are used as a mechanism to distribute work across threads or machines. Commonly, even a small microservice would require to run tasks in the background (async). Hence, this boilerplate comes with a pre-loaded popular task queue Celery.

The configuration is quite simple as we are using Flask's factory pattern. We have also added a separate customized logger for celery logs which writes to a separate configurable file celery.log. Running celery is as easy as just running a command celery worker -A celery_worker.celery -l=info (from the root of our project).

9. Deployment

Flask Documentation provides a very good list of deployment options and the necessary steps as well. Also, you can refer to my earlier post on Deploying the Django application which is similar for flask application as well (with some minute changes). 😄

Deploying Django with Celery and Redis on Ubuntu

Idris Rampurawala ・ Nov 5 '19

#devops #django #python #webdev

10. API Caching and Validation

We already covered a lot of points but two main aspects of having a robust microservice are to have an efficient validation and caching capability. This boilerplate comes with following two packages to achieve the same:

Redis	A Python Redis client for caching
Webargs	A Python library for parsing and validating HTTP request objects

🤓 Getting started with flask boilerplate

So far we talked about most of the points an ideal project requires that comes pre-loaded with this boilerplate. Whatever may be your use case for creating an app such as an email service, a communication service, reporting service, or even a microservice, you can use this boilerplate with minimal customization.

You can start using flask boilerplate using the following installation steps and kick start your project!

# clone the repo
$ git clone https://github.com/idris-rampurawala/flask-boilerplate.git
# move to the project folder
$ cd flask-boilerplate
# create pipenv environment for python 3
$ pipenv --three
# activate the pipenv environment
$ pipenv shell
# install all dependencies (include -d for installing dev dependencies)
$ pipenv install -d
# create `.env` from `.env.example` and set appropriate environment variables
# run the project
$ python run.py
# optionally check if everything is working correctly by calling status apps (check Test section of readme)

All set! Flask boilerplate is up and running and waiting for you to write some cool stuff from it 🥳

⭐ Useful Links

If you find this helpful or have any suggestions, feel free to comment. Also, do not forget to hit ❤️ or 🦄 if you like my post.

See ya! until my next post 😋

Creating a Search Filter in Angular

Idris Rampurawala — Thu, 07 May 2020 12:16:53 +0000

Consider a scenario where we have a long list of data being displayed to the user on UI. It would be cumbersome for the user to search for any particular keyword in this long list without any search functionality provided. Hence, to make our users' life easy, we would usually implement search filters on our UI.

So now the question is, how to implement it? It's quite easy though 😉 All we want is a filter that takes an array as input and returns a subset of that array based on the term we supply. In Angular, this way of transforming data to some other form is implemented with Pipes. Let's first understand a bit more about pipes before we start the implementation.

Pipes in Angular

A pipe takes in data as input and transforms it into the desired output. A pipe can be used in both the HTML template expression and in a component. Angular does provide us with some built-in pipes such as CurrencyPipe, DatePipe, DecimalPipe, etc. Check this code snippet below to see it in action.



dateObj = Date.now();

// HTML template expression syntax using pipe operator (|)
{{ dateObj | date }}               // output is 'Jun 15, 2015'
{{ dateObj | date:'medium' }}      // output is 'Jun 15, 2015, 9:43:11 PM'
{{ dateObj | date:'shortTime' }}   // output is '9:43 PM'
{{ dateObj | date:'mm:ss' }}       // output is '43:11'

// Using in component
constructor(private datePipe: DatePipe) { 
    console.log(datePipe.transform(Date.now(),'yyyy-MM-dd'));
    //2019-07-22
}

Pipes are of 2 types - Pure and Impure. For more information on Angular pipes, visit this link.

Implementing Search Filter

1. Create the Filter Pipe

Let's populate the pipe with code for the filter. Copy and paste this code into filter.pipe.ts:



// filter.pipe.ts

import { Pipe, PipeTransform } from '@angular/core';

@Pipe({ name: 'appFilter' })
export class FilterPipe implements PipeTransform {
  /**
   * Pipe filters the list of elements based on the search text provided
   *
   * @param items list of elements to search in
   * @param searchText search string
   * @returns list of elements filtered by search text or []
   */
  transform(items: any[], searchText: string): any[] {
    if (!items) {
      return [];
    }
    if (!searchText) {
      return items;
    }
    searchText = searchText.toLocaleLowerCase();

    return items.filter(it => {
      return it.toLocaleLowerCase().includes(searchText);
    });
  }
}

This pipe definition reveals the following key points:

A pipe is a class decorated with pipe metadata.
The pipe class implements the PipeTransform interface's transform method that accepts an input value followed by optional parameters and returns the -transformed value. In our filter pipe, it takes 2 inputs - an array and the search text to filter the array with.
To tell Angular that this is a pipe, we apply the @Pipe decorator, which we import from the core Angular library.
The @Pipe decorator allows us to define the pipe name that we'll use within template expressions. It must be a valid JavaScript identifier. Our pipe's name is appFilter.

2. Using Pipe

To use the pipe, first, we need to import it into the app module. Our app.module.ts file would now look like this:



// app.module.ts

import { BrowserModule } from '@angular/platform-browser';
import { NgModule } from '@angular/core';
import { FormsModule } from '@angular/forms';

import { AppComponent } from './app.component';

import { FilterPipe } from './pipes/filter.pipe'; // -> imported filter pipe

@NgModule({
  declarations: [
    AppComponent,
    FilterPipe // -> added filter pipe to use it inside the component
  ],
  imports: [
    BrowserModule,
    FormsModule
  ],
  bootstrap: [AppComponent]
})
export class AppModule { }

Now we can use the filter pipe in our App Component. Let's assume that in our app.component.html we have an input box where we can enter our searchText and a list that makes use of this pipe to filter the results.



<!-- app.component.html -->

<div class="content" role="main">
  <div class="card">
    <div class="form-group">
      <label for="search-text">Search Text</label>
      <input type="email" class="form-control" id="search-text" aria-describedby="search-text" 
        [(ngModel)]="searchText" placeholder="Enter text to search" 
        autofocus>
    </div>
    <ul class="list-group list-group-flush">
      <!-- results of ngFor is passed to appFilter with argument searchText -->
      <li class="list-group-item" *ngFor="let c of characters | appFilter: searchText">
        {{c}}
      </li>
    </ul>
  </div>
</div>



// app.component.ts

import { Component } from '@angular/core';

@Component({
  selector: 'app-root',
  templateUrl: './app.component.html',
  styleUrls: ['./app.component.scss']
})
export class AppComponent {
  title = 'angular-text-search-highlight';
  searchText = '';
  characters = [
    'Ant-Man',
    'Aquaman',
    'Asterix',
    'The Atom',
    'The Avengers',
    'Batgirl',
    'Batman',
    'Batwoman',
    ...
  ]
}

That's it! Now when we run our app, we will see the following output:

But hey! our search results are not being highlighted as it was shown at the beginning 😟

The reason is that Pipes in angular only transforms the data passed to it into the desired output. It won't manipulate the HTML associated with it. To highlight the search results, we would be required to manipulate the HTML to highlight the searchText part of it. This can be achieved using Directives.

Directives in Angular

Angular directives are used to extend the power of the HTML by giving it new syntax. There are 3 types of directives:

Components — directives with a template.
Structural directives — change the DOM layout by adding and removing DOM elements.
Attribute directives — change the appearance or behavior of an element, component, or another directive.

Covering directives is outside the scope of this post. If you want to learn more about angular directives, visit this link.

Implementing Directive in our Application

In our case, we will be using the attribute directive to highlight the searchText in the result list.

1. Creating highlight directive

An attribute directive minimally requires building a controller class annotated with @Directive, which specifies the selector that identifies the attribute. The controller class implements the desired directive behavior.

Let's populate the directive with code for the highlighting. Copy and paste this code into highlight.pipe.ts:



// highlight.directive.ts

import { Directive, Input, SimpleChanges, Renderer2, ElementRef, OnChanges } from '@angular/core';

@Directive({
  selector: '[appHighlight]'
})
export class HighlightDirective implements OnChanges {
  @Input() searchedWord: string; // searchText
  @Input() content: string; // HTML content
  @Input() classToApply: string; //class to apply for highlighting
  @Input() setTitle = false; //sets title attribute of HTML

  constructor(private el: ElementRef, private renderer: Renderer2) { }

  ngOnChanges(changes: SimpleChanges): void {
    if (!this.content) {
      return;
    }

    if (this.setTitle) {
      this.renderer.setProperty(
        this.el.nativeElement,
        'title',
        this.content
      );
    }

    if (!this.searchedWord || !this.searchedWord.length || !this.classToApply) {
      this.renderer.setProperty(this.el.nativeElement, 'innerHTML', this.content);
      return;
    }

    this.renderer.setProperty(
      this.el.nativeElement,
      'innerHTML',
      this.getFormattedText()
    );
  }

  getFormattedText() {
    const re = new RegExp(`(${this.searchedWord})`, 'gi');
    return this.content.replace(re, `<span class="${this.classToApply}">$1</span>`);
  }
}

The logic is to manipulate the current HTML element by adding <span> tag in between the searchText and applying the highlighting class to it.

2. Using directive

To use the pipe, first, we need to import it into the app module. Our app.module.ts file would now look like this:



// app.module.ts

import { BrowserModule } from '@angular/platform-browser';
import { NgModule } from '@angular/core';
import { FormsModule } from '@angular/forms';

import { AppComponent } from './app.component';

import { HighlightDirective } from './directives/highlight.directive'; // ->  imported directive
import { FilterPipe } from './pipes/filter.pipe';

@NgModule({
  declarations: [
    AppComponent,
    HighlightDirective, // -> added directive
    FilterPipe
  ],
  imports: [
    BrowserModule,
    FormsModule
  ],
  providers: [],
  bootstrap: [AppComponent]
})
export class AppModule { }

To use this directive in our HTML file, we will add it as a normal HTML attribute with all its parameters. It would look like this:



<!-- app.component.html -->

<div class="content" role="main">
  <div class="card">
    <div class="form-group">
      <label for="search-text">Search Text</label>
      <input type="email" class="form-control" id="search-text" aria-describedby="search-text" 
        [(ngModel)]="searchText" placeholder="Enter text to search" 
        autofocus>
    </div>
    <ul class="list-group list-group-flush">
      <li class="list-group-item" *ngFor="let c of characters | appFilter: searchText"
        appHighlight [searchedWord]="searchText" [content]="c"  
        [classToApply]="'font-weight-bold'" [setTitle]="'true'">
        {{c}}
      </li>
    </ul>
  </div>
</div>

Now, we would be able to see the desired output! 😌

You check out my GitHub repo for a complete implementation of this post.

See ya! until my next post 😋

Share Your AWS S3 Private Content With Others, Without Making It Public

Idris Rampurawala — Tue, 24 Mar 2020 12:26:33 +0000

Amazon Web Services (AWS) S3 objects are private by default. Only the object owner has permission to access these objects. Optionally we can set bucket policy to whitelist some accounts or URLs to access the objects of our S3 bucket.

Recently, while working on a project, I came across a scenario where I wanted to make objects of my bucket public but only to limited users. Consider some of the scenarios as follows:

You store some user-specific files on S3 which is accessible to your respective authenticated users via your frontend. But what if a user wants to share this file with the people outside of your web application user accounts (who are not part of your application or simply unauthorized users)?
You have some user reports stored on the S3 bucket. And you want to email these reports to your users and some other email ids that have been allowed by the user? An email has a limit for the size of the attachment and hence not feasible to attach all reports in one email.

So how can we allow these S3 objects to be accessed by public users (users not part of our system)? The answer is presigned url. The objects can be shared with others by creating a presigned url, to grant time-limited permission to download (view) the objects. Anyone who receives the presigned url can then access the object. For example, if you have a video in your bucket and both the bucket and the object are private, you can share the video with others by generating a presigned url.

Out of all possible ways to generate a presigned url of an S3 object, let's discuss the 2 most important ones:

S3 presigned url method
Amazon CloudFront's signed url method

✔️ Prerequisite

Before we begin, we are assuming that we have an S3 bucket idr-aws-bucket with an object test.gif on AWS.
This bucket is private i.e. when we try to access the object we get the following error.
We are gonna use boto3 - AWS SDK for Python for code demonstration. For any other language, the process is the same but the code would defer. Links provided at the end of the post.

Let's roll in!🍟

1. S3 presigned url method

A user who does not have AWS credentials or permission to access an S3 object can be granted temporary access by using a presigned url.

A presigned url is generated by an AWS user who has access to the object. The generated url is then given to the user without making our bucket private. The presigned url can be entered in a browser or used by a program or HTML webpage. The credentials used by the presigned url are those of the AWS user who generated the url.

This method is pretty easy but has got some points to consider. Let's implement this and then look into the shortcomings of using this method.

Step 1. Configuring IAM user credentials required for AWS SDK

Anyone with valid security credentials can create a presigned url. However, to successfully access an object, the presigned url must be created by someone who has permission to perform the operation that the presigned url is based upon. Yes, you read that correctly! You can perform any operation such as GetObeject, PutObject, etc while generating the presigned url (will see later in code below).

Step 2. Setup AWS SDK

Once we have our AWS credentials by which we gonna create presigned url, we need to set up AWS SDK to access the method. You can find the documentation here.

Step 3. The Code



from typing import Optional

# import the AWS library
import boto3
from botocore.exceptions import ClientError


def create_presigned_url(
        bucket_name: str, object_name: str, expiration=3600) -> Optional[str]:
    """Generate a presigned URL to share an s3 object

    Arguments:
        bucket_name {str} -- Required. s3 bucket of object to share
        object_name {str} -- Required. s3 object to share

    Keyword Arguments:
        expiration {int} -- Expiration in seconds (default: {3600})

    Returns:
        Optional[str] -- Presigned url of s3 object. If error, returns None.
    """

    # Generate a presigned URL for the S3 object
    aws_profile = <your-aws-profile-name>
    s3_client = boto3.session.Session(
        profile_name=aws_profile).client('s3')
    try:
        # note that we are passing get_object as the operation to perform
        response = s3_client.generate_presigned_url('get_object',
                                                    Params={
                                                        'Bucket': bucket_name,
                                                        'Key': object_name
                                                    },
                                                    ExpiresIn=expiration)
    except ClientError as e:
        logging.error(e)
        return None

    # The response contains the presigned URL
    return response

This is a wrapper function that returns the presigned url when called with an S3 bucket and object name. Also, note that the expiration can also be set which allows the url to remain valid up to expiration (in seconds).



# function that calls above function to create presigned url
def generate_presigned_url():
    bucket_name = <your-s3-bucket-name>
    bucket_resource_url = <your-s3-bucket-resource-url>
    url = create_presigned_url(
        bucket_name,
        bucket_resource_url
    )
    return {
        'url': url
    }



// Output of above function
{
  "url": "https://<your-aws-bucket>.s3.amazonaws.com/test.gif?AWSAccessKeyId=AKIA5UAF6XOYXYXLIIVG&Signature=ztJYSThTOmpqU4nVSertoNuNYLE%3D&Expires=1584881770"
}

👇 Points to remember

This method will allow our object to be accessible via the generated url and hence, the presigned url is the actual object path of our AWS S3 bucket, which is now public in the url.
This method will not modify any bucket policy or any other bucket permissions. It will be validated at the time we visit the url.
Presigned url is signed by the credentials we are using to create it and hence any change in the url will fail the authentication resulting in an unauthorized request by AWS.
The credentials that we can use to create a presigned url include:
- AWS Identity and Access Management (IAM) instance profile: Token can be valid up to a maximum of 6 hours.
- AWS Security Token Service: Token can be valid up to a maximum of 36 hours when signed with permanent credentials, such as the credentials of the AWS account root user or an IAM user.
- IAM user: Token can be valid up to a maximum of 7 days when using AWS Signature Version 4.

Hence, by this method, we can create presigned url valid up to a maximum of 7 days. For higher expiration value, we need to move to the second approach i.e. by using CloudFront

2. Amazon CloudFront's signed url method

We have looked into how easy it is to share our S3 object with the outside world using the AWS SDK. Though it is relatively easy with no extra setup in AWS, there are limitations:

Our S3 object url is now visible to the person we shared presigned url with
URL can be valid for a maximum of 7 days (if created via IAM user credentials) Hence, for more flexibility, AWS CloudFront is the second approach to go for. It is not that easy but if you stick to the post, we will get it there. ☺️

Let us understand how it works before implementation:

AWS CloudFront provides 2 methods to allow users to access our private content - signed urls or signed cookies. For simplicity, we will look into the implementation via signed urls.
Firstly, we will generate CloudFront key-pairs for an AWS account to allow the account to create a signed url. For simplicity, I am using my root account for the same. Optionally, you can create a user for the same and add the account id of the user in the AWS CloudFront distribution set up process below.
Next, we will have to create the AWS CloudFront distribution to allow our bucket to be accessible via AWS CloudFront signed url.
Finally, we will write the code to use the private key created in step 1 above to create signed url for an S3 object.

1. Generating CloudFront key-pairs

Remember, I am creating a key-pair for my root IAM user here. Click on the My Security Credentials option.

Next, as highlighted in the image below, click on the button Create New Key Pair

This will open up a pop-up where you can download both your private and public keys and also make a note of your access ID as well.

2. Creating CloudFront Distribution

Create CloudFront Distribution
On the Cloudfront dashboard, click on Create Distribution.
Select a content delivery method
As we are gonna access our S3 content via the web (HTTP), we will choose the method web (as highlighted in the image).
Add distribution settings
In the Origin Settings section, we will select an Amazon S3 bucket that we want to make available for users via signed url, and make sure you select rest of the options as in the image below:

📕 Two important points to note here, I have enabled Restrict Viewer Access and in Trusted Signers I have checked Self. Trusted Signers are the AWS Accounts which are authorized to create signed urls. Checking the Self option means that I am using my AWS root IAM user (for which we create key-pair in step 1) to generate the signed url. Rest all of the settings are self-explanatory.
Make a note of your distribution domain

3. Finally, the code to create signed url 😌



from datetime import datetime

from botocore.exceptions import ClientError
from botocore.signers import CloudFrontSigner
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import padding

def rsa_signer(message: str) -> str:
    # cloudfront-pk.pem is the private key generated in step 1 for IAM user
    with open('.cloudfront-pk.pem', 'rb') as key_file:
        private_key = serialization.load_pem_private_key(
            key_file.read(),
            password=None,
            backend=default_backend()
        )
    return private_key.sign(
        message, padding.PKCS1v15(), hashes.SHA1())


def create_cloudfront_signed_url(
        object_name: str, expiration_date: datetime) -> str:
    # cloudfront key-par access ID generated in step 1
    key_id = <your-cloudfront-key-par-access-ID>
    # your cloudfront distribution domain created in step 4 of distribution creation steps
    cloudfront_domain = <your-cloudfront-distribution-domain>

    url = '{cloudfront_domain}/{object_name}'.format(
        cloudfront_domain=cloudfront_domain,
        object_name=object_name
    )

    cloudfront_signer = CloudFrontSigner(key_id, rsa_signer)

    # Create a signed url that will be valid until the specfic expiry date
    # provided using a canned policy.
    signed_url = cloudfront_signer.generate_presigned_url(
        url, date_less_than=expiration_date)
    return signed_url

Once you execute the above function, you will get the out as shown in the image below.



from datetime import datetime, timedelta

# function that calls above function to create presigned url
def generate_presigned_url():
    expire_date = datetime.utcnow() + timedelta(days=2) # expires in 2 days
    bucket_resource_url = <your S3 bucket resource url>
    url = create_cloudfront_signed_url(
        bucket_resource_url,
        expire_date
    )
    return {
        'url': url
    }



// Output of above function
{
  "url": "https://<your-cloudfront-id>.cloudfront.net/test.gif?Expires=1585148785&Signature=AeRhPIEAfWX4g-5rpfMkyQD1-pjNdUj2sUG8yyL-mBEcYG6Vhk2jCTpQMJ2m6oUkz-6zZXSqT-3v2KKB-HSHqG984ZBq8rKe~ZWMC3k3I3gqW9x8u-S-Xl20Fj6CyydunW6pDWILxyLZeU5s2x2aYZok2PKyns0eUgBpls0oPb6jX3WVdRafmIIU98EZXEqx5mTWi0se62cqDrI8MPk5Em7gGJ8vhh5X7PNebMEVQcqf189Yj~3RSPxTwIKylzRc6ZyMU5wq8pGeKkrwZK~iQO5DXqHRXOFzI8JM5XL7608F7aFl6uSlboMvKZWEh0FGqmp8GZpR45sbUOKT38AJ3g__&Key-Pair-Id=<key-pair-id>"
}

📢 Final Words

Congratulations! 👏 We have successfully managed to solve one of the key challenges in AWS of serving private S3 content to users without making it public with the help of signed url. Do not forget to check the Useful Links section to find links to read more about this topic.

Also, I have created a small Flask project on GitHub demonstrating the code to implement both the approaches we have talked about here.

idris-rampurawala / aws-s3-access-demo

This project is an implementation to show various approaches to share your AWS S3 private object with others without making it public.

AWS S3 Object Access Demo

Share Your AWS S3 Private Object With Others Without Making It Public

This project is an implementation to show various approaches to share your AWS S3 private object with others without making it public. You can find an in-depth article on this implementation here.

Background

There are various instances where we want to share our S3 object with users temporarily or with some specific expiration time without the need to make our S3 bucket private. This project aims to solve that problem by creating presigned url with some code examples.

Prerequisites

Python 3.7.2 or higher
Install pip
AWS account with an S3 bucket and…

View on GitHub

⭐ Useful Links

My GitHub repository implementing both of the above approaches
S3 presigned url links
CloudFront signed url links

If you find this helpful or have any suggestions, feel free to comment. Also, do not forget to hit ❤️ or 🦄 if you like my post.

See ya! until my next post 😋

DEV Community: Idris Rampurawala

Building a Scalable API Event Logger using Pub/Sub, and BigQuery

☑️ Prerequisite

📇 Overview

What does "event-based API logging" mean?

What are some limitations of "old-style" logging?

🧠 API Logging Flow

⚙️ The Logging Middleware

The APILogEvent struct

The Middleware

🧩 Integration with BigQuery

🤔 Why Not Just Use GCP Structured Logs?

📊 What We Can Do with the Data

🚀 Benefits

idris-rampurawala / api-pubsub-logger

Demo Go API with middleware that logs requests/responses to Google Cloud Pub/Sub. Features versioned routes, sensitive data masking, request tracking, and comprehensive unit tests.

API Pub/Sub Logger

Blog Post

Overview

Features

📑 References

Implementing a Redis-Based Task Queue with Configurable Concurrency

☑️ Prerequisite

Overview

1. queue_task_executor

2. queue_task_executor_callback

💡 Advantages of This Redis Queue Setup vs. Cloud Infrastructure

📢 Final Words

idris-rampurawala / redis-queue

Redis Queue project is aimed at creating a queue based in redis and celery using redis data structures.

Redis Queue

Contributing

Getting Started

Prerequisites

Project setup

Creating virtual environment

Handling Duplicates in BigQuery

1. Deleting the duplicates before a re-run

2. Selecting the unique records

🔹 Logical Views vs Materialized Views

🔹 Logical View Syntax

📑 References

Underrated Skills of a Senior Software Engineer

👁️‍🗨️ Attention To Detail

🛠️ Optimize Dev Processes

⚖️ Minimize Over-engineering

🔎 Problem Discovery

📄 Requirement Focused

💬 Conclusion

Writing Custom Django Database Functions

📜 Django Func(*expressions, **extra) class

🔹 Custom database functions

🔹 Partial implementation of database functions

🔹 Executing a GROUP BY without an aggregation function

📑 Resources

Parallelize Processing a Large AWS S3 File

1. Find the total bytes of the S3 file

2. Create a celery task to process a chunk

3. Execute multiple celery tasks in parallel

🔍 Comparing the processing time

✔️ Benefits of this approach

idris-rampurawala / s3-select-demo

This project showcases the rich AWS S3 Select feature to stream a large data file in a paginated style.

AWS S3 Select Demo

Background

📑 Resources

Efficiently Streaming a Large AWS S3 File via S3 Select

S3 Select

🧱 Constructing SQL expressions

🌫️ Streaming Chunks

1. Find the total bytes of the S3 file

2. Create a generator to stream the chunks

idris-rampurawala / s3-select-demo

This project showcases the rich AWS S3 Select feature to stream a large data file in a paginated style.

AWS S3 Select Demo

Background

✔️ Benefits of using S3-Select

❗ Limitations of S3 Select

📑 Resources

Parallelize Processing a Large AWS S3 File

1. `queue_task_executor`

2. `queue_task_executor_callback`

📜 Django `Func(*expressions, **extra)` class

🔹 Executing a `GROUP BY` without an aggregation function