DEV Community: John Owolabi Idogun

Integrating Stripe with SvelteKit: Dynamic Pricing and Payment Security

John Owolabi Idogun — Wed, 28 May 2025 14:40:59 +0000

Introduction

Integrating Stripe into a SvelteKit application, especially when dealing with dynamic product pricing and robust webhook validation, can be challenging. While building a client's business application using SvelteKit and PostgreSQL (with Drizzle ORM), I encountered the need for a streamlined integration with Stripe. The goal was to use minimal dependencies while supporting features such as dynamic pricing, product images, and secure payment verification via webhooks, all without requiring database updates. This article provides a comprehensive guide to this entire process, addressing a gap in existing resources. It should be noted that we cover the concepts here, and as a result, it's framework agnostic. You can use NextJS or NuxtJS in place of SvelteKit since they support server endpoints, form actions, etc.

Live version

Source code

Sirneij / digibooks

A simple demo app that integrates Stripe with SvelteKit

Project Specification: Simple Digital Books App with SvelteKit & Stripe

1. Project Goal

To build a minimalist digital books application using SvelteKit to demonstrate Stripe integration for purchasing digital goods. The application will allow users to browse books, add them to a cart, and purchase them using Stripe. Access to purchased content will be granted based on the email provided during checkout.

2. Core Features

2.1. Book Catalog

Display a list of available digital books.
Each book entry should show:
- Title
- Author
- Cover Image (URL)
- Price
Allow users to view more details for a single book (e.g., a longer description).

2.2. Shopping Cart

Users can add books from the catalog to a shopping cart.
Users can view the contents of their shopping cart.
Users can remove books from their shopping cart.
The cart should display the subtotal and total price.

2.3. Stripe Checkout

Initiate a Stripe Checkout session for the…

View on GitHub

System architecture and requirements

To demonstrate this integration, we will build a minimalist digital book application. The application will allow users to browse books, add them to a cart, and purchase them using Stripe. Access to purchased content will be granted based on the email provided during checkout.

Prerequisite

You need to create a new SvelteKit application. If you do not have one yet, just follow these commands:

npx sv create digibooks

When prompted during npx sv create digibooks:

Choose SvelteKit minimal (barebones scaffolding for your new app) when asked Which Svelte app template?.
Select your preferred options for other features. For this project, I opted for TypeScript syntax, Prettier, ESLint, Tailwind CSS, and Drizzle ORM (with libSQL).

Also, we will be using the stripe library, which should be installed:

npm install stripe

Implementation

Step 1: Create the database schema

When you opt for drizzle ORM while creating the project with npx sv create, src/lib/server/db/index.ts (with a .env entry for DATABASE_URL), src/lib/server/db/schema.ts and drizzle.config.ts alongside db:push, db:migrate and db:studio entries in package.json will be made available. Let's open src/lib/server/db/schema.ts and populate it with:

import { sql } from "drizzle-orm";
import { sqliteTable, integer, text } from "drizzle-orm/sqlite-core";

export const books = sqliteTable("books", {
  id: integer("id").primaryKey({ autoIncrement: true }),
  title: text("title").notNull(),
  author: text("author").notNull(),
  description: text("description"),
  coverImageUrl: text("cover_image_url"),
  // Store price in cents to avoid floating point issues
  priceInCents: integer("price_in_cents").notNull(),
});

export const purchases = sqliteTable("purchases", {
  id: integer("id").primaryKey({ autoIncrement: true }),
  quantity: integer("quantity").notNull().default(1),
  customerEmail: text("customer_email").notNull(),
  bookId: integer("book_id")
    .notNull()
    .references(() => books.id, { onDelete: "cascade" }), // Optional: onDelete behavior
  purchasedAt: integer("purchased_at", { mode: "timestamp" })
    .notNull()
    .default(sql`(strftime('%s', 'now'))`),
  stripeCheckoutSessionId: text("stripe_checkout_session_id").notNull(),
  isCompleted: integer("is_completed", { mode: "boolean" })
    .notNull()
    .default(false), // Default to false, indicating the purchase is not completed
});

Just some very basic SQL tables to list books and record purchases linked to customer emails, which can then be used to grant access to the digital content.

To effect this change, run:

npm run db:push

and choose the Yes when prompted. Now, our database tables have been created. We will also add a simple "money" formatter that takes into consideration the cents-based pricing in src/lib/utils/helpers.ts:

/**
 * Format a number as currency
 *
 * @param {number | string | null | undefined} amount - Amount to format in cents
 * @returns {string} - Formatted currency string
 */
export const formatMoney = (
  amount: number | string | null | undefined
): string => {
  if (amount === null || amount === undefined || isNaN(Number(amount))) {
    return "$0.00"; // Default to $0.00 if input is invalid
  }
  const validAmount = (Number(amount) || 0) / 100; // Convert cents to dollars
  return validAmount.toLocaleString("en-US", {
    style: "currency",
    currency: "USD",
    currencyDisplay: "narrowSymbol",
  });
};

To support a Turso-hosted SQLite instance, we need to modify src/lib/server/db/index.ts and drizzle.config.ts to avoid this error:

LibsqlError: SERVER_ERROR: Server returned HTTP status 401
    at mapHranaError (~/digibooks/node_modules/@libsql/client/lib-esm/hrana.js:279:16)
    at file:///Users/johnidogun/Documents/projects/digibooks/node_modules/@libsql/client/lib-esm/http.js:76:23
    at process.processTicksAndRejections (node:internal/process/task_queues:105:5)
    at async Object.query (~/digibooks/node_modules/drizzle-kit/bin.cjs:79275:25)
    at async fromDatabase3 (~/digibooks/node_modules/drizzle-kit/bin.cjs:20507:23) {
  code: 'SERVER_ERROR',
  rawCode: undefined,
  [cause]: HttpServerError: Server returned HTTP status 401
      at errorFromResponse (~/digibooks/node_modules/@libsql/hrana-client/lib-esm/http/stream.js:352:16)
      at process.processTicksAndRejections (node:internal/process/task_queues:105:5) {
    status: 401
  }
}

Open up drizzle.config.ts and make it look like this:

import { defineConfig } from "drizzle-kit";

if (!process.env.DATABASE_URL) throw new Error("DATABASE_URL is not set");
if (!process.env.DATABASE_AUTH_TOKEN)
  throw new Error("DATABASE_AUTH_TOKEN is not set");

export default defineConfig({
  schema: "./src/lib/server/db/schema.ts",
  dialect: process.env.DATABASE_URL?.startsWith("file:") ? "sqlite" : "turso",
  dbCredentials: {
    url: process.env.DATABASE_URL,
    authToken: process.env.DATABASE_AUTH_TOKEN,
  },
  verbose: true,
  strict: true,
});

and src/lib/server/db/index.ts:

import { drizzle } from "drizzle-orm/libsql";
import { createClient } from "@libsql/client";
import * as schema from "./schema";
import { env } from "$env/dynamic/private";

if (!env.DATABASE_URL) throw new Error("DATABASE_URL is not set");
if (!env.DATABASE_AUTH_TOKEN) throw new Error("DATABASE_AUTH_TOKEN is not set");

const client = createClient({
  url: env.DATABASE_URL,
  authToken: env.DATABASE_URL.startsWith("file:")
    ? undefined
    : env.DATABASE_AUTH_TOKEN,
});

export const db = drizzle(client, { schema });

We are conditionally (based on the DATABASE_URL) changing the database dialect to either sqlite or turso. If turso and authToken are required to successfully connect to the instance. The dialect in drizzle.config.ts is conditionally set to turso when a non-file DATABASE_URL is used, which allows dbCredentials to correctly utilize the authToken. If you are not using a Turso SQLite instance, you may not need to worry about this.

Seeding the database

So that we can have enough data to work with, this article's GitHub repository contains two files: books.json and seed.ts. The former contains some dummy data in JSON, while the latter has the code to load (not optimized) this data into the DB. I also added an entry in the script section of package.json where we utilize the experimental node's --experimental-transform-types flag to run the TypeScript file. You can run these commands in succession (locally) after deploying or at the start of development of your app:

npm run db:push # handles everything database migration
npm run db:seed # runs the seeding logic

Step 2: Cart and carting store

This app will not deal with the intricacies of authentication, so there will not be "users". However, we still need a way to keep track of what the current user wants to buy. As a result, we need a carting system that won't need to store its data in the database. We will combine the browser's localStorage and Svelte 5 runes to achieve a reliable reactivity. Create a src/lib/states/carts.svelte.ts and populate it with:

import { browser } from "$app/environment";
import type { Book } from "$lib/server/db/schema";
import type { CartItem } from "$lib/types/cart";
import { formatMoney } from "$lib/utils/helpers";

// Load cart from localStorage if available
function loadCartFromStorage(): CartItem[] {
  if (!browser) return [];
  try {
    const saved = localStorage.getItem("digibooks-cart");
    return saved ? JSON.parse(saved) : [];
  } catch {
    return [];
  }
}

// Save cart to localStorage
function saveCartToStorage(cart: CartItem[]): void {
  if (!browser) return;
  try {
    localStorage.setItem("digibooks-cart", JSON.stringify(cart));
  } catch (error) {
    console.error("Failed to save cart to localStorage:", error);
  }
}

// Cart state management using Svelte 5 runes
function createCartState() {
  let items = $state<CartItem[]>(loadCartFromStorage());

  function addItem(book: Book) {
    const existingItem = items.find((item) => item.book.id === book.id);

    if (existingItem) {
      existingItem.quantity += 1;
    } else {
      items.push({ book, quantity: 1 });
    }

    saveCartToStorage(items);
  }

  function removeItem(bookId: number) {
    items = items.filter((item) => item.book.id !== bookId);
    saveCartToStorage(items);
  }

  function updateQuantity(bookId: number, quantity: number) {
    if (quantity <= 0) {
      removeItem(bookId);
      return;
    }

    const item = items.find((item) => item.book.id === bookId);
    if (item) {
      item.quantity = quantity;
      saveCartToStorage(items);
    }
  }
  function clear() {
    items = [];
    saveCartToStorage(items);
  }

  function getTotalItems(): number {
    return items.reduce((total, item) => total + item.quantity, 0);
  }

  function getTotalPrice(): number {
    return items.reduce(
      (total, item) => total + item.book.priceInCents * item.quantity,
      0
    );
  }

  function getFormattedTotal(): string {
    return formatMoney(getTotalPrice());
  }

  return {
    get items() {
      return items;
    },
    addItem,
    removeItem,
    updateQuantity,
    clear,
    getTotalItems,
    getTotalPrice,
    getFormattedTotal,
  };
}

// Create a singleton instance
export const cartState = createCartState();

This is a basic CRUD carting process while making items reactive using the power of Svelte 5 rune. We referenced a type here, and this is the definition:

import type { Book } from '$lib/server/db/schema';

// Cart item interface
export interface CartItem {
    book: Book;
    quantity: number;
}

export interface SessionMetadata {
    customerEmail?: string;
    userId?: string; // Optional user ID for authenticated users
    timestamp: string; // ISO string for when the session was created
    books: string; // Array of book IDs and their quantities
    itemCount?: string; // Number of items in the cart
    totalAmount?: string; // Total amount in cents as a string
}

export interface PaginationMetadata {
    page: number;
    limit: number;
    total: number;
    totalPages: number;
}
=

Just some additional types that will be used later. Now, we can add a simple utility function (not extremely necessary) for the addToCart functionality:

...
/**
 * Add a book to the cart
 *
 * @param {Book} book - The book to add to the cart
 */
export const addToCart = (book: Book) => {
    cartState.addItem(book);
};

Step 3: Data loading and frontend

Though we will not delve into the intricacies of styles and tailwindcss stuff as those are not the focus of this article, beware that I made custom changes to the app's src/app.css, src/routes/+layout.svelte and of course, each of the pages: src/routes/+page.svelte, src/routes/[id]/+page.svelte,src/routes/cart/+page.svelte, and src/routes/purchases/+page.svelte with some glasmorphism effects.

Let's see what src/routes/+page.server.ts looks like:

import { db } from "$lib/server/db";
import { books } from "$lib/server/db/schema";
import { or, type SQL, like, asc, desc, count } from "drizzle-orm";
import type { PageServerLoad } from "./$types";

export const load: PageServerLoad = async ({ url }) => {
  const searchQuery = url.searchParams.get("search");
  const sortBy = url.searchParams.get("sort") || "featured";
  const page = parseInt(url.searchParams.get("page") || "1");
  const limit = 8;
  const offset = (page - 1) * limit;

  const whereClause: SQL[] = [];
  // Apply search filter
  if (searchQuery) {
    whereClause.push(
      like(books.title, `%${searchQuery}%`),
      like(books.author, `%${searchQuery}%`)
    );
  }

  const orderByClause: SQL[] = [];

  // Apply sorting
  switch (sortBy) {
    case "price-low":
      orderByClause.push(asc(books.priceInCents));
      break;
    case "price-high":
      orderByClause.push(desc(books.priceInCents));
      break;
    case "title":
      orderByClause.push(asc(books.title));
      break;
    case "author":
      orderByClause.push(asc(books.author));
      break;
    default: // featured
      orderByClause.push(asc(books.id));
      break;
  }

  // Execute both queries in parallel
  const [countResult, allBooks] = await Promise.all([
    // Count query
    db
      .select({ count: count() })
      .from(books)
      .where(whereClause.length > 0 ? or(...whereClause) : undefined)
      .then((result) => result[0]),

    // Books query with pagination
    db
      .select()
      .from(books)
      .where(whereClause.length > 0 ? or(...whereClause) : undefined)
      .orderBy(...orderByClause)
      .limit(limit)
      .offset(offset),
  ]);

  return {
    books: allBooks || [],
    pagination: {
      total: countResult.count,
      page,
      limit,
      totalPages: Math.ceil(countResult.count / limit),
    },
    searchQuery,
    sortBy,
  };
};

Svelte automatically generates $types:

SvelteKit automatically generates TypeScript definitions for your routes, including the data shapes for load functions and page props. When you see import type { PageServerLoad } from "./$types"; or use PageData (often imported from ./$types as well, or inferred), these types are derived from your load function's return signature and any route parameters. This provides excellent type safety between your server-side data loading and your client-side Svelte components. If you modify the data structure returned by a load function, SvelteKit's type generation will reflect these changes, helping you catch errors at build time.

It simply fetches data from the DB with search, filtering, and pagination support. To prevent request waterfalls, we use Promise.all to run the counting and data retrieval queries in parallel. This can improve performance. In src/routes/+page.svelte, we simply get this data via the data props and render them, excluding other features:

<script lang="ts">
    import BookCard from '$lib/components/BookCard.svelte';
    import Pagination from '$lib/components/utils/Pagination.svelte';
  import { addToCart } from '$lib/utils/helpers';
  ...

    let { data }: { data: PageData } = $props();
    ...
</script>

<svelte:head>
    <title>Books - DigiBooks</title>
</svelte:head>

<div class="space-y-8">
    <!-- Clean Header -->
    <div class="text-center">
        <h1 class="text-content mb-3 text-4xl font-bold">Digital Books</h1>
        <p class="text-content-muted">Discover your next great read</p>
    </div>
    ...
    <!-- Books Grid/List -->
    {#if data.books.length > 0}
        <div
            class="grid gap-6 transition-all duration-300"
            class:grid-cols-1={viewMode === 'list'}
            class:sm:grid-cols-2={viewMode === 'grid'}
            class:lg:grid-cols-3={viewMode === 'grid'}
            class:xl:grid-cols-4={viewMode === 'grid'}
        >
            {#each data.books as book, index (book.id)}
                <div in:fly={{ y: 20, duration: 300, delay: index * 30 }}>
                    <BookCard {book} {viewMode} addToCart={() => addToCart(book)} />
                </div>
            {/each}
        </div>
    ...

    <!-- Pagination -->
    {#if data.pagination.totalPages > 1}
        <Pagination metadata={data.pagination} />
    {/if}
</div>

Refer to the repository for the full code and the BookCard and Pagination components. Same goes with src/routes/[id]/+page.svelte where its load function is quite basic:

import { error } from "@sveltejs/kit";
import type { PageServerLoad } from "./$types";
import { db } from "$lib/server/db";
import { books } from "$lib/server/db/schema";
import { eq } from "drizzle-orm";

export const load: PageServerLoad = async ({ params }) => {
  const book = await db
    .select()
    .from(books)
    .where(eq(books.id, Number(params.id)))
    .get();

  if (!book) {
    error(404, "Book not found");
  }
  return {
    book,
  };
};

Then comes the /purchases route, which provides an interface for users to see the products they purchased. It requires that users input their email address. This is the page's +page.server.ts:

import { db } from "$lib/server/db";
import { purchases, books } from "$lib/server/db/schema";
import { eq, and, desc } from "drizzle-orm";
import type { PageServerLoad } from "./$types";

export const load: PageServerLoad = async ({ url }) => {
  const email = url.searchParams.get("email");

  if (!email) {
    return {
      purchases: [],
      email: null,
    };
  }

  try {
    // Fetch purchases with book details for the given email
    const userPurchases = await db
      .select({
        id: purchases.id,
        quantity: purchases.quantity,
        customerEmail: purchases.customerEmail,
        purchasedAt: purchases.purchasedAt,
        stripeCheckoutSessionId: purchases.stripeCheckoutSessionId,
        isCompleted: purchases.isCompleted,
        book: {
          id: books.id,
          title: books.title,
          author: books.author,
          description: books.description,
          coverImageUrl: books.coverImageUrl,
          priceInCents: books.priceInCents,
        },
      })
      .from(purchases)
      .innerJoin(books, eq(purchases.bookId, books.id))
      .where(
        and(eq(purchases.customerEmail, email), eq(purchases.isCompleted, true))
      )
      .orderBy(desc(purchases.purchasedAt));
    return {
      purchases: userPurchases,
      email,
    };
  } catch (error) {
    console.error("Failed to fetch purchases:", error);
    return {
      purchases: [],
      email,
      error: "Failed to load purchases",
    };
  }
};

To reduce trips to the Database, we joined the purchases and books. Its frontend simply displays this data.

In the next subsection, we will finally integrate Stripe!

Step 4: Payment integration

It's now time to integrate Stripe with our app so that users can securely pay for their books of choice. There are multiple ways to integrate Stripe or accept payments with Stripe:

Prebuilt Checkout page: Stripe hosts the payment page. You redirect users to Stripe, and Stripe redirects them back to your site after payment. This is the quickest way to get started and ensures PCI compliance is handled by Stripe.
Payment Element: Embeddable UI components that allow you to design a custom payment form on your site while still leveraging Stripe's infrastructure for processing and PCI compliance. Offers more customization than the prebuilt Checkout page.
Custom payment flow: Build your entire payment UI from scratch and use Stripe APIs (like Stripe.js on the frontend and the Stripe SDK on the backend) to process payments. This offers maximum control over the user experience but also requires more effort to implement and manage PCI compliance.

In this article, we'll go with the first option. However, there is a plan to extend to the other two in future articles (as well as integrate other payment processors such as Squareup and Braintree).

To start, create a payments.ts file in src/lib/server/payments.ts (you can use server in NextJS):

import { STRIPE_SECRET_KEY } from "$env/static/private";
import type { CartItem, SessionMetadata } from "$lib/types/cart";
import Stripe from "stripe";

export const stripe = new Stripe(STRIPE_SECRET_KEY);

export async function createCheckoutSession({
  items,
  successUrl,
  cancelUrl,
  customerEmail,
  metadata,
}: {
  items: CartItem[];
  successUrl: string;
  cancelUrl: string;
  customerEmail?: string;
  metadata?: SessionMetadata;
}) {
  try {
    // Convert items to Stripe line items
    const lineItems = items.map((item) => ({
      price_data: {
        currency: "usd",
        product_data: {
          name: item.book.title,
          description: item.book.description || undefined,
          images: item.book.coverImageUrl
            ? [item.book.coverImageUrl]
            : undefined,
          metadata: {
            bookId: item.book.id,
          },
        },
        unit_amount: item.book.priceInCents,
      },
      quantity: item.quantity,
    }));
    const sessionParams: Stripe.Checkout.SessionCreateParams = {
      line_items: lineItems,
      mode: "payment",
      success_url: successUrl,
      cancel_url: cancelUrl,
      billing_address_collection: "required",
      payment_intent_data: {
        metadata: {
          ...metadata,
          source: "digibooks",
        },
      },
      metadata: {
        ...metadata,
      },
    };

    // Add customer email if provided
    if (customerEmail) {
      sessionParams.customer_email = customerEmail;
    }

    const session = await stripe.checkout.sessions.create(sessionParams);

    return {
      success: true,
      sessionId: session.id,
      url: session.url,
    };
  } catch (error) {
    console.error("Stripe checkout session creation failed:", error);

    return {
      success: false,
      error:
        error instanceof Error
          ? error.message
          : "Failed to create checkout session",
    };
  }
}

To instantiate a Stripe instance, a STRIPE_SECRET_KEY is required — gotten from your Stripe developer dashboard. In this function, we only touched on some of the data accepted by a Stripe checkout session. You can provide much more data depending on your needs. In the code, we supplied the products (books) being paid for alongside their images and descriptions with the unit amount (in cents, stripe takes cents for USD and EUR. It takes whatever is the $\frac{1}{100}$ of your supported currency. This is to avoid floating point issues). The mode is a one-time payment. You could use subscription for recurring payments or setup to charge customers later. We also provided the routes for successful and unsuccessful processing. We even made collecting customers' billing addresses mandatory. After a payment has been confirmed (via a webhook, to be implemented later), we need some data about the books and other things the user bought so we can fullfil them (here create purchases entry(ies)), that's a potential use case of Stripe.Checkout.SessionCreateParams.metadata. It only takes primitives (string, number, null) as objects' values, though that's why the books property in SessionMetadata is a string (JSON string of an array of bookId and quantity). This is just the tip of the iceberg. You can do much more. After supplying these data, Stripe internally sends requests to its API(s), and a checkout session identification is returned alongside other important details such as the section URL (where you should redirect users to for them to make payment). Here is a truncated example response:

{
  "id": "cs_test_a11YYufWQzNY63zpQ6QSNRQhkUpVph4WRmzW0zWJO2znZKdVujZ0N0S22u",
  "object": "checkout.session",
  ...
  "amount_subtotal": 2198,
  "amount_total": 2198,
  "automatic_tax": {
    "enabled": false,
    "liability": null,
    "status": null
  },
  "billing_address_collection": null,
  "cancel_url": null,
  ...
  "created": 1679600215,
  "currency": "usd",
  ...
  "customer_email": null,
  ...
  ...
  "metadata": {},
  "mode": "payment",
  "payment_intent": null,
  "payment_link": null,
  ...
  "success_url": "https://example.com/success",
  "total_details": {
    "amount_discount": 0,
    "amount_shipping": 0,
    "amount_tax": 0
  },
  "url": "https://checkout.stripe.com/c/pay/cs_test_a11YYufWQzNY63zpQ6QSNRQhkUpVph4WRmzW0zWJO2znZKdVujZ0N0S22u#fidkdWxOYHwnPyd1blpxYHZxWjA0SDdPUW5JbmFMck1wMmx9N2BLZjFEfGRUNWhqTmJ%2FM2F8bUA2SDRySkFdUV81T1BSV0YxcWJcTUJcYW5rSzN3dzBLPUE0TzRKTTxzNFBjPWZEX1NKSkxpNTVjRjN8VHE0YicpJ2N3amhWYHdzYHcnP3F3cGApJ2lkfGpwcVF8dWAnPyd2bGtiaWBabHFgaCcpJ2BrZGdpYFVpZGZgbWppYWB3dic%2FcXdwYHgl"
}

With that, we can now implement the form action that uses this function:

import { createCheckoutSession } from "$lib/server/payments";
import type { CartItem } from "$lib/types/cart";
import { fail, redirect } from "@sveltejs/kit";
import type { Actions } from "./$types";

export const actions: Actions = {
  checkout: async ({ request, url }) => {
    const formData = await request.formData();
    const cartItems = JSON.parse(
      formData.get("cartItems") as string
    ) as CartItem[];
    const customerEmail = formData.get("email") as string;

    // Validate cart items
    if (!cartItems || !Array.isArray(cartItems) || cartItems.length === 0) {
      return fail(400, {
        error: "Cart is empty",
      });
    }

    const result = await createCheckoutSession({
      items: cartItems,
      customerEmail: customerEmail || undefined,
      metadata: {
        userId: "guest", // Replace with actual user ID if authenticated
        timestamp: new Date().toISOString(),
        books: JSON.stringify(
          cartItems.map((item) => ({
            bookId: item.book.id.toString(),
            quantity: item.quantity,
          }))
        ),
        itemCount: cartItems
          .reduce((total, item) => total + item.quantity, 0)
          .toString(),
        totalAmount: cartItems
          .reduce(
            (total, item) => total + item.book.priceInCents * item.quantity,
            0
          )
          .toString(),
      },
      successUrl: `${url.origin}/cart?checkout=success`,
      cancelUrl: `${url.origin}/cart?checkout=cancel`,
    });

    if (!result.success) {
      return fail(500, {
        error: result.error || "Failed to create checkout session",
      });
    }

    // Redirect to Stripe checkout
    redirect(303, result.url!);
  },
};

We simply retrieve the needed data from the form, slightly process it, and send it to Stripe.

Beware of putting a redirect in try...catch:

Wrapping a redirect in try...catch in SvelteKit form action almost always returns the `catch` block (with errors). You should handle errors using a different approach.

Now, let's see the cart page (only shows the form and URL handling part):

<script lang="ts">
    import { formatMoney } from '$lib/utils/helpers';
    ...

    let isLoading = $state(false),
        email = $state(''),
        redirectCountdown = $state(0),
        isRedirecting = $state(false);

    // Check for checkout status in URL params
    const checkoutStatus = $derived(page.url.searchParams.get('checkout'));

    function updateQuantity(bookId: number, quantity: number) {
        cartState.updateQuantity(bookId, quantity);
    }

    function removeItem(bookId: number) {
        cartState.removeItem(bookId);
    }

    function clearCart() {
        cartState.clear();
    }

    // Handle checkout success/cancel
    onMount(() => {
        if (checkoutStatus === 'success') {
            // Clear cart on successful checkout
            cartState.clear();

            ...
        }
    });
</script>

<svelte:head>
    <title>Cart - DigiBooks</title>
</svelte:head>

<div class="mx-auto max-w-6xl">
    <!-- Header -->
    <div class="mb-8">
        <h1 class="mb-2 text-3xl font-bold text-gray-900">Shopping Cart</h1>
        <p class="text-gray-600">
            {cartState.items.length === 0
                ? 'Your cart is empty'
                : `${cartState.getTotalItems()} ${cartState.getTotalItems() === 1 ? 'item' : 'items'}`}
        </p>
    </div>

    <!-- Checkout Status Messages -->
    {#if checkoutStatus === 'success'}
        <div class="mb-6 rounded-lg border border-green-200 bg-green-50 p-6" in:fade>
            <div class="flex items-start">
                {@render check({ class: 'mr-3 mt-0.5 h-6 w-6 text-green-400', 'aria-hidden': 'true' })}
                <div class="flex-1">
                    <h3 class="mb-2 font-medium text-green-800">Payment Successful!</h3>
                    <p class="mb-3 text-green-700">
                        Thank you for your purchase. Your books have been added to your library.
                    </p>
                    {#if isRedirecting}
                        <div class="flex items-center gap-2 text-sm text-green-600">
                            {@render loader({ class: 'h-4 w-4 animate-spin' })}
                            <span>Redirecting to your purchases in {redirectCountdown} seconds...</span>
                        </div>
                        <button
                            onclick={() => goto('/purchases')}
                            class="mt-3 text-sm font-medium text-green-600 underline hover:text-green-500"
                        >
                            Go to purchases now
                        </button>
                    {/if}
                </div>
            </div>
        </div>
    {:else if checkoutStatus === 'cancel'}
        <div class="mb-6 rounded-lg border border-yellow-200 bg-yellow-50 p-4" in:fade>
            <div class="flex items-center">
                {@render alert({ class: 'mr-2 h-5 w-5 text-yellow-400', 'aria-hidden': 'true' })}
                <p class="font-medium text-yellow-800">
                    Checkout was cancelled. Your items are still in your cart.
                </p>
            </div>
        </div>
    {/if}

    <!-- Form Error -->
    {#if page.form?.error}
        ...
    {/if}

    {#if cartState.items.length > 0}
        <div class="grid grid-cols-1 gap-8 lg:grid-cols-12">
            <!-- Cart Items -->
            <div class="lg:col-span-8">
                <div class="overflow-hidden rounded-2xl border border-gray-200 bg-white">
                    <!-- Items Header -->
                    <div class="flex items-center justify-between border-b border-gray-200 px-6 py-4">
                        <h2 class="text-lg font-semibold text-gray-900">Items</h2>
                        <button onclick={clearCart} class="text-sm font-medium text-red-600 hover:text-red-700">
                            Clear all
                        </button>
                    </div>

                    <!-- Items List -->
                    <div class="divide-y divide-gray-200">
                        {#each cartState.items as item, index (item.book.id)}
                            ...
                        {/each}
                    </div>
                </div>
            </div>

            <!-- Order Summary -->
            <div class="lg:col-span-4">
                <div class="sticky top-8 rounded-2xl border border-gray-200 bg-white p-6">
                    <h2 class="mb-6 text-lg font-semibold text-gray-900">Order Summary</h2>

                    <!-- Summary Details -->
                    <div class="mb-6 space-y-4">
                        <div class="flex justify-between text-gray-600">
                            <span>Subtotal ({cartState.getTotalItems()} items)</span>
                            <span>{formatMoney(cartState.getTotalPrice())}</span>
                        </div>
                        <div class="flex justify-between text-gray-600">
                            <span>Shipping</span>
                            <span class="font-medium text-green-600">Free</span>
                        </div>
                        <div class="flex justify-between text-gray-600">
                            <span>Tax</span>
                            <span>$0.00</span>
                        </div>
                        <hr class="border-gray-200" />
                        <div class="flex justify-between text-lg font-bold text-gray-900">
                            <span>Total</span>
                            <span class="text-blue-600">{cartState.getFormattedTotal()}</span>
                        </div>
                    </div>

                    <!-- Checkout Form -->
                    <form
                        method="POST"
                        action="?/checkout"
                        use:enhance={() => {
                            isLoading = true;
                            return async ({ update, result }) => {
                                await update();
                                isLoading = false;
                                if (result.type === 'redirect') {
                                    window.location.href = result.location;
                                }
                            };
                        }}
                        class="space-y-4"
                    >
                        <!-- Hidden cart items -->
                        <input type="hidden" name="cartItems" value={JSON.stringify(cartState.items)} />

                        <!-- Email Input -->
                        <div>
                            <label for="email" class="mb-2 block text-sm font-medium text-gray-700">
                                Email Address
                            </label>
                            <input
                                type="email"
                                id="email"
                                name="email"
                                bind:value={email}
                                required
                                disabled={isLoading}
                                class="w-full rounded-lg border border-gray-300 px-3 py-2 text-sm focus:border-blue-500 focus:ring-1 focus:ring-blue-500 focus:outline-none disabled:bg-gray-50 disabled:opacity-50"
                                placeholder="your@email.com"
                            />
                        </div>
                        <!-- Action Buttons -->
                        <div class="space-y-3">
                            <button
                                type="submit"
                                disabled={isLoading || cartState.items.length === 0 || !email.trim()}
                                class="w-full rounded-xl bg-blue-600 px-4 py-3 font-medium text-white transition-all duration-200 hover:bg-blue-700 focus:ring-2 focus:ring-blue-500 focus:ring-offset-2 disabled:cursor-not-allowed disabled:bg-gray-400"
                            >
                                {#if isLoading}
                                    <div class="flex items-center justify-center gap-2">
                                        {@render loader({ class: 'h-5 w-5 animate-spin' })}
                                        <span>Processing...</span>
                                    </div>
                                {:else}
                                    Proceed to Checkout
                                {/if}
                            </button>

                            <a
                                href="/"
                                class="block w-full rounded-xl bg-gray-100 px-4 py-3 text-center font-medium text-gray-800 transition-colors hover:bg-gray-200"
                            >
                                Continue Shopping
                            </a>
                        </div>
                    </form>

                    <!-- Security Badge -->
                    <div class="mt-6 flex items-center justify-center gap-2 text-sm text-gray-500">
                        {@render shield({ class: 'h-5 w-5 text-blue-600' })}
                        <span>Secure checkout guaranteed</span>
                    </div>
                </div>
            </div>
        </div>
    {:else}
        ...
    {/if}
</div>

Just some styles and progressive form enhancements.

Step 5: Confirming purchases with webhook

Currently, even after a successful payment, users cannot access what they paid for. This is unfair. While we strive to fulfill every payment made, we need to be careful, though, as people can game the payment flow, providing a fraudulent payment. This is why we didn't create any purchases entry at the point of making payments. One way we could fulfill an order is to manually create the purchase entries from our Stripe dashboard. This is "manual" and hence prone to errors and very tedious. There is an automatic way to do this with Stripe, and it is called webhook — a mechanism that allows one application to notify another application in real-time about specific events. Before we can fulfill orders, we want Stripe to notify us that payments have indeed been made (via its checkout.session.completed or checkout.session.async_payment_succeeded hooks). We need to patiently listen to this, and a good way is via an endpoint. To locally test this, kindly look into this stripe guide. You will need to create a destination (the url of your endpoint where you want stripe to send communications to, for this app, it'll be /api/webhooks/stripe) and you will be prompted to select the hooks you want to listen to as well (I opted for only checkout.session.completed). After this, a webhook secret will be generated for you. Copy it and save it as STRIPE_WEBHOOK_SECRET in your environment variable. Locally, it will be generated when you first run stripe listen --events checkout.session.completed,checkout.session.async_payment_succeeded --forward-to localhost:5173/api/webhooks/stripe (you have a lot of events to listen to!). To "trigger" an event locally, just run stripe trigger checkout.session.completed in another terminal while the listen command is running in another. Your app should also be running. Here is our endpoint code:

import { STRIPE_WEBHOOK_SECRET } from "$env/static/private";
import { db } from "$lib/server/db";
import { purchases } from "$lib/server/db/schema";
import { stripe } from "$lib/server/payments";
import type { SessionMetadata } from "$lib/types/cart.js";
import { error, json } from "@sveltejs/kit";
import { eq, and } from "drizzle-orm";

export async function POST({ request }) {
  const body = await request.text();
  const signature = request.headers.get("stripe-signature");

  if (!signature) {
    error(400, "Missing stripe-signature header");
  }

  try {
    const event = stripe.webhooks.constructEvent(
      body,
      signature,
      STRIPE_WEBHOOK_SECRET
    );

    if (
      event.type === "checkout.session.completed" ||
      event.type === "checkout.session.async_payment_succeeded"
    ) {
      const session = event.data.object;
      const metadata = session.metadata as unknown as SessionMetadata;

      if (!metadata.books || !metadata.timestamp) {
        error(400, "Missing required metadata");
      }

      const items = JSON.parse(metadata.books) as Array<{
        bookId: string;
        quantity: number;
      }>;
      const customerEmail =
        session.customer_details?.email || session.customer_email || "unknown";

      try {
        const purchaseRecords = [];

        // Check each item individually to avoid duplicates from webhook retries
        for (const item of items) {
          const existingPurchase = await db
            .select()
            .from(purchases)
            .where(
              and(
                eq(purchases.stripeCheckoutSessionId, session.id),
                eq(purchases.bookId, parseInt(item.bookId))
              )
            )
            .limit(1);

          if (existingPurchase.length === 0) {
            purchaseRecords.push({
              quantity: item.quantity,
              customerEmail,
              bookId: parseInt(item.bookId),
              stripeCheckoutSessionId: session.id,
              isCompleted: true,
            });
          }
        }

        if (purchaseRecords.length > 0) {
          await db.insert(purchases).values(purchaseRecords);
          console.log(
            `✅ Created ${purchaseRecords.length} new purchase records for session ${session.id}`
          );
        } else {
          console.log(
            `ℹ️ All purchases already exist for session ${session.id}`
          );
        }
      } catch (dbError) {
        console.error("Database error creating purchases:", dbError);
        error(500, "Failed to create purchase records");
      }
    }

    return json({ received: true });
  } catch (err) {
    console.error("Webhook error:", err);
    error(400, "Webhook Error");
  }
}

To avoid fraudulent event notifications, we ensure the event is signed and verified. Unsigned and/or unverified events are outrightly rejected. After that, the event is constructed, and since we are only interested in checkout.session.completed (covers most successful one-time payments) and async_payment_succeeded (important for asynchronous payment methods (e.g., some bank transfers)) type, that's what we listen to. From there, we retrieve the details we sent previously during payment (from metadata) and from it create purchases entries. This is done in an idempotent way so that webhook retries by Stripe or anything won't create duplicates, which would result in losing money.

That's it! Thank you for your time!

Outro

Enjoyed this article? I'm a Software Engineer and Technical Writer actively seeking new opportunities to impact and learn, particularly in areas related to web security, finance, healthcare, and education. If you think my expertise aligns with your team's needs, let's chat! You can find me on LinkedIn and X. I am also an email away.

Quantum Information Processing: Foundations - Part 1

John Owolabi Idogun — Wed, 23 Apr 2025 14:17:50 +0000

For better formatting of math, citations, and all, it is recommended to visit https://johnowolabiidogun.dev/blog/quantum-information-processing-foundations-part-1-9e816e/6801dc9d6e2495ac3e5db09c. Also, part 2 is already posted there.

Introduction

For decades, computation has mostly been carried out via machines that represent data in one of two states: $0$ or $1$ . These machines, most probably including the one you are reading this with, are classified as classical computers. The rise in the study of quantum mechanics and information theory in the last decades of the twentieth century birthed a more powerful alternative [@Rieffel2011] termed Quantum Computing (QC), a term commonly attributed to Richard Feynman and the independent work of Yuri Manin [@Rieffel2011]. QC extends classical computing's data representation to include "superposed" (linear combinations) states — which can be infinite [@Bernhardt2019]. This idea of superposition is the cornerstone of this computing paradigm. And instead of classical computers' $bi t$ , quantum computers' base unit is $q u bi t$ — an abbreviation of quantum and bit, coined by Ben Schumacher in his 1995 "Quantum coding" paper [@Schumacher1995]. In simple terms, QC fundamentally, using Mathematics and Physics, supercharges computing. As demonstrated by Deutsch-Jozsa, Simon, Grover, and Shor's algorithms (which will be discussed in this series), we can have speedups ranging from quadratic to exponential in computational algorithms. Certain encryption protocols have, as reported in October 2024, been broken by Chinese researchers using a quantum algorithm [@Swayne2024].

In this series, we'll be introduced to the theories of QC from notations, measurements, superposition, and entanglement to sophisticated algorithms such as Grover's and Shor's. We will use worked mathematical examples accompanied (mostly) by code verification (using Qiskit and/or Cirq). Most of the problems will be taken from [@Rieffel2011], and their mathematical solutions will be more thoroughly explained for clarity. This is to really understand the concepts. I suggest you check out [@Rieffel2011,@Bernhardt2019] for deeper or more theoretically approachable knowledge.

Prerequisite

Quantum computing relies on mathematical principles, but you can learn the essentials without being a math whiz. A working knowledge of high school math will equip you to understand the applications and fundamental ideas. Familiarity with the Python programming language will be helpful to understand qiskit and/or cirq code.

Dirac notation

In quantum mechanics/physics, Dirac notation, named after Paul Adrien Maurice Dirac, the English Mathematician and Theoretical Physicist who developed it in the 1930s, is used to depict quantum states alongside their transformations [@Rieffel2011]. It is composed of two vectors: $∣ψ⟩\ket{\psi}$ and $⟨ψ∣\bra{\psi}$ . $∣ψ⟩\ket{\psi}$ , termed $k e t$ , represents the column vector of the quantum states whereas $⟨ψ∣\bra{\psi}$ , regarded as $b r a$ , is the row vector.

Important things to note here are:

$ψ†\psi^\dagger$ is equivalent to the $b r a$ , $⟨ψ∣\bra{\psi}$ , and is obtained by taking the Hermitian (conjugate transpose) of $ψ\psi$ . For example, if:

i]\psi = \frac{1}{\sqrt{2}} \begin{bmatrix} 1 \ i \end{bmatrix}

Then its conjugate transpose (bra) is:

ψ†=⟨ψ∣=12[1−i]\psi^\dagger = \bra{\psi} = \frac{1}{\sqrt{2}} \begin{bmatrix} 1 & -i \end{bmatrix}

This operation involves taking the transpose and then applying complex conjugation to each element [@Lopez2025]. It can then easily be inferred that:

\psi \cdot \psi^\dagger = \left( \frac{1}{\sqrt{2}} \begin{bmatrix} 1 \ i \end{bmatrix} \right) \cdot \left( \frac{1}{\sqrt{2}} \begin{bmatrix} 1 & -i \end{bmatrix} \right)

\frac{1}{2} \begin{bmatrix} 1 \ i \end{bmatrix} \begin{bmatrix} 1 & -i \end{bmatrix} = \frac{1}{2} \begin{bmatrix} 1 \cdot 1 & 1 \cdot (-i) \ i \cdot 1 & i \cdot (-i) \end{bmatrix} = \frac{1}{2} \begin{bmatrix} 1 & -i \ i & 1 \end{bmatrix}

Note that $ψ⋅ψ†≠I\psi \cdot \psi^\dagger \neq I$ , but instead produces a projection matrix. If we instead compute:

\psi^\dagger \cdot \psi = \left( \frac{1}{\sqrt{2}} \begin{bmatrix} 1 & -i \end{bmatrix} \right) \cdot \left( \frac{1}{\sqrt{2}} \begin{bmatrix} 1 \ i \end{bmatrix} \right)

\frac{1}{2} \left( 1 \cdot 1 + (-i) \cdot i \right) = \frac{1}{2} \left( 1 + 1 \right) = 1

So we conclude that the inner product (braket) is $1$ :

⟨ψ∣ψ⟩=1\braket{\psi|\psi} = 1

However, if $ψ\psi$ is $u ni t a ry$ — a matrix whose inverse is equal to its conjugate transpose — then:

ψ†⋅ψ=ψ⋅ψ†=I\psi^\dagger \cdot \psi = \psi \cdot \psi^\dagger = I

For example, consider the Hadamard gate $H$ :

\frac{1}{\sqrt{2}} \begin{bmatrix} 1 & 1 \ 1 & -1 \end{bmatrix}

Its Hermitian conjugate is:

1−1]H^\dagger = H^T = \frac{1}{\sqrt{2}} \begin{bmatrix} 1 & 1 \ 1 & -1 \end{bmatrix}

Since all entries are real. Then:

H^\dagger H = \frac{1}{2} \begin{bmatrix} 1 & 1 \ 1 & -1 \end{bmatrix} \begin{bmatrix} 1 & 1 \ 1 & -1 \end{bmatrix} = \frac{1}{2} \begin{bmatrix} 1+1 & 1-1 \ 1-1 & 1+1 \end{bmatrix} = \begin{bmatrix} 1 & 0 \ 0 & 1 \end{bmatrix} = I

Therefore, $H$ is unitary. All quantum gates (to be discussed) are unitary.

While the inner product results in a number (scalar), the outer product, $∣ψ⟩⟨ψ∣\ket{\psi}\bra{\psi}$ , produces a matrix:

\ket{\psi}\bra{\psi} = \frac{1}{2} \begin{bmatrix} 1 \ i \end{bmatrix} \begin{bmatrix} 1 & -i \end{bmatrix} = \frac{1}{2} \begin{bmatrix} 1 & -i \ i & 1 \end{bmatrix}

This outer product is a nifty tool for converting matrices to Dirac notation.

In a Hilbert space (quantum vector space), states like $∣0⟩\ket{0}$ and $∣1⟩\ket{1}$ are:
- Orthogonal: $⟨0∣1⟩=0\braket{0|1} = 0$
- Normalized: $⟨0∣0⟩=1\braket{0|0} = 1$

Superposition Principle and Measurements

In quantum mechanics, as in general physics, a system remains in an indeterminate state until it is measured. Quantum states are typically represented in superpositions, following the superposition principle, which states [@Shor2025]:

Let $∣a⟩\ket{a}$ and $∣b⟩\ket{b}$ be two quantum states that are perfectly distinguishable — that is, orthogonal. If $α\alpha$ and $β\beta$ are complex numbers such that the sum of their squared magnitudes is $1$ , then the linear combination (superposition):

α∣a⟩+β∣b⟩\alpha \ket{a} + \beta \ket{b}

is a valid quantum state.

An important takeaway here is the condition that "the sum of their squared magnitudes is 1". This is the simplified consequence of the Born rule, formulated by the German-British physicist Max Born in his 1926 paper [@Born1926].

In QC, measurements are more nuanced, as there are multiple possible measurement bases. The most commonly used is the computational basis, consisting of the states
$∣0⟩\ket{0}$ and $∣1⟩\ket{1}$ . Let's go through examples (Exercise 2.6 in [@Rieffel2011]) of how to measure in QC.

Question 1:

Describe the possible measurement outcome and give the probability for the outcome for this pair consisting of a state and a measurement basis:

\ket{\psi} = \frac{\sqrt{3}}{2}\ket{0} + \frac{1}{2}\ket{1}, {\ket{0}, \ket{1}}

Solution:

When a quantum state is subjected to a measurement in a specific basis, the possible results of that measurement directly correspond to the states that constitute the measurement basis. In this case, since the measurement basis is $∣0⟩,∣1⟩{\ket{0}, \ket{1}}$ , the possible outcomes of measuring the state $∣ψ⟩\ket{\psi}$ are obtaining the state $∣0⟩\ket{0}$ or obtaining the state $∣1⟩\ket{1}$ .

Now, let's proceed to estimate its probabilities:

\mathcal{P}_{\ket{0}} = \left|\frac{\sqrt{3}}{2}\right|^{2} = \frac{3}{4}

\mathcal{P}_{\ket{1}} = \left|\frac{1}{2}\right|^{2} = \frac{1}{4}

Check:

Recall that from the Born rule [@Born1926], a state:

∣ψ⟩=α∣0⟩+β∣1⟩\ket{\psi} = \alpha\ket{0} + \beta\ket{1}

must have its probabilities (squares of the amplitudes) equal to $1$ :

∣α∣2+∣β∣2=1\left|\alpha\right|^{2} + \left|\beta\right|^{2} = 1

\mathcal{P}{\ket{0}} + \mathcal{P}{\ket{1}} = \frac{3}{4} + \frac{1}{4} = \frac{1 + 3}{4} = 1

Code implementation using Google `cirq`

To begin, we need to define a $q u bi t$ in Cirq. This can be done using cirq.NamedQubit or cirq.LineQubit. For this example, we will use cirq.NamedQubit. Next, we create a quantum circuit using cirq.Circuit(). To prepare the desired quantum state $∣ψ⟩=32∣0⟩+12∣1⟩\ket{\psi} = \frac{\sqrt{3}}{2}\ket{0} + \frac{1}{2}\ket{1}$ , we will define the target state vector with numpy.array and pass its result in cirq.StatePreparationChannel.

There is another, more mathematical and accurate way to do this, using rotation, but since we haven't introduced the concept yet, we will go with this.

Then, to verify the probabilities, we need to add a measurement operation to the circuit in the computational basis. Cirq provides the cirq.measure() function or the cirq.MeasurementGate for this purpose. To retrieve the results, we need to simulate the quantum circuit. cirq.Simulator() shines here. The complete code is now:

import cirq
import numpy as np

# Define a qubit
q = cirq.NamedQubit('q')

# Create a circuit
circuit = cirq.Circuit()

# Define the target state vector
target_state = np.array([np.sqrt(3)/2, 1/2])

# Prepare the state using StatePreparationChannel
circuit.append(cirq.StatePreparationChannel(target_state).on(q))

# Add a measurement operation
circuit.append(cirq.measure(q, key='result'))

print("Circuit to prepare the state:")
print(circuit)

# Create a simulator
simulator = cirq.Simulator()

# Run the simulation multiple times
repetitions = 1000
result = simulator.run(circuit, repetitions=repetitions)

# Get the measurement results
measurement_counts = result.histogram(key='result')
print(f"\nMeasurement counts after {repetitions} repetitions:")
print(measurement_counts)

# Calculate the probabilities from the simulation
probability_0 = measurement_counts.get(0, 0) / repetitions
probability_1 = measurement_counts.get(1, 0) / repetitions

print(f"\nSimulated probability of outcome 0: {probability_0:.3f}")
print(f"Simulated probability of outcome 1: {probability_1:.3f}")

Question 2:

Describe the possible measurement outcome and give the probability for the outcome for this pair consisting of a state and a measurement basis:

∣ψ⟩=∣−i⟩,∣0⟩,∣1⟩\ket{\psi} = \ket{-i}, {\ket{0}, \ket{1}}

Solution

The measurement basis for this is exactly equal to the previous one. It's only the state that differs. So the possible outcomes still hold.

Let's zoom in on the state:

∣ψ⟩=∣−i⟩\ket{\psi} = \ket{-i}

We'll introduce the concept of the Bloch Sphere here.

The Bloch Sphere is a geometric representation of a qubit's state space, which helps visualize superposition.

Here's an illustration:

{Figure 1: Bloch Sphere showing the major measurement basis.}

According to the Bloch Sphere (will be discussed more later) above, the given state:

∣−i⟩=12(∣0⟩−i∣1⟩)=12∣0⟩−12i∣1⟩\ket{-i} = \frac{1}{\sqrt{2}}\left( \ket{0} - i\ket{1} \right) = \frac{1}{\sqrt{2}}\ket{0} -\frac{1}{\sqrt{2}}i\ket{1}

From this, their probabilities are calculated as follows:

\mathcal{P}_{\ket{0}} = \left| \frac{1}{\sqrt{2}} \right|^{2} = \frac{1}{2}

\mathcal{P}_{\ket{1}} = \left| -\frac{1}{\sqrt{2}}i \right|^{2} = \left| - i \right|^{2} \times \left| \frac{1}{\sqrt{2}} \right|^{2} = - \times -1 \times \frac{1}{2} = \frac{1}{2}

Since $i2=(−1)2=−1i^2 = (\sqrt{-1})^2= -1$ .

Check:

$P∣0⟩+P∣1⟩=12+12=1\mathcal{P}{\ket{0}} + \mathcal{P}{\ket{1}} = \frac{1}{2} + \frac{1}{2} = 1$ .

Simulating the circuit with Qiskit or Cirq is left as an exercise to the reader.

Question 3:

Describe the possible measurement outcome and give the probability for the outcome for this pair consisting of a state and a measurement basis:

∣ψ⟩=∣0⟩,∣+⟩,∣−⟩\ket{\psi} = \ket{0}, {\ket{+}, \ket{-}}

Solution

As mentioned earlier, when a quantum system is measured with respect to a particular basis, the possible outcomes of that measurement are the basis states themselves. In this scenario, the measurement is performed in the Hadamard basis, which is comprised of the states $∣+⟩\ket{+}$ and $∣−⟩\ket{-}$ . Consequently, the only possible outcomes of measuring a qubit in the ${+}, \ket{-}}$ basis are the qubit collapsing into either the $∣+⟩\ket{+}$ state or the $∣−⟩\ket{-}$ state.

To find the probabilities of these outcomes, we need to know that:

∣+⟩≡∣0⟩+∣1⟩2\ket{+} \equiv \frac{\ket{0} + \ket{1}}{\sqrt{2}}

∣−⟩≡∣0⟩−∣1⟩2\ket{-} \equiv \frac{\ket{0} - \ket{1}}{\sqrt{2}}

and that the given state, $∣0⟩\ket{0}$ , is:

∣0⟩=12(∣+⟩+∣−⟩)\ket{0} = \frac{1}{\sqrt{2}}\left(\ket{+} + \ket{-} \right)

From here, we can simply estimate the probabilities of each of the outcomes:

\mathcal{P}_{\ket{+}} = \left| \frac{1}{\sqrt{2}} \right|^{2} = \frac{1}{2}

\mathcal{P}_{\ket{-}} = \left| \frac{1}{\sqrt{2}} \right|^{2} = \frac{1}{2}

Another approach (which works for even the preceding and succeeding exercises since it's fundamental) is to use the idea of inner product. Since the measurement basis is ${+}, \ket{-}}$ and the given state is $∣0⟩\ket{0}$ , the probability of each measurement outcome is the square of the inner product of the state with the measurement outcome. In other words:

P∣+⟩=∣⟨+∣0⟩∣2\mathcal{P}{\ket{+}} = \left|\braket{+|0}\right|^2

P∣−⟩=∣⟨−∣0⟩∣2\mathcal{P}{\ket{-}} = \left|\braket{-|0}\right|^2

$∴\therefore$

\mathcal{P}_{\ket{+}} = \left| \braket{\frac{1}{\sqrt{2}}(\ket{0} + \ket{1})|0} \right|^2 = \frac{1}{2}\left| (\bra{0} + \bra{1})\ket{0} \right|^2

The $b r a$ simply transforms $∣0⟩\ket{0}$ and $∣1⟩\ket{1}$ to row vectors: $⟨0∣\bra{0}$ and $⟨1∣\bra{1}$ . Multiplying out:

\mathcal{P}_{\ket{+}} = \frac{1}{2}\left| \braket{0|0} + \braket{1|0} \right|^2 = \frac{1}{2}|1+0|^2 = \frac{1}{2}

Similarly,

\mathcal{P}_{\ket{-}} = \left| \braket{\frac{1}{\sqrt{2}}(\ket{0} - \ket{1})|0} \right|^2 = \frac{1}{2}\left| (\bra{0} - \bra{1})\ket{0} \right|^2

\frac{1}{2}\left| \braket{0|0} - \braket{1|0} \right|^2 = \frac{1}{2}|1-0|^2 = \frac{1}{2}

Check

This is another cirq code for this:

import cirq
import numpy as np

# Define a qubit
qubit = cirq.LineQubit(0)

# Create a quantum circuit
circuit = cirq.Circuit()

# The qubit is initialized to |0>

# Apply a Hadamard gate to prepare for measurement in the Hadamard basis
circuit.append(cirq.H(qubit))

# Measure the qubit in the computational basis
circuit.append(cirq.measure(qubit, key='result'))

# Print the circuit
print("Quantum Circuit:")
print(circuit)

# Simulate the circuit many times to get probabilities
simulator = cirq.Simulator()
num_simulations = 1000
results = simulator.run(circuit, repetitions=num_simulations)

# Get the measurement results
measurement_outcomes = results.measurements['result']

# Count the occurrences of each outcome (0 and 1)
counts = np.unique(measurement_outcomes, return_counts=True)
print(f'{counts[0][0]}: {counts[1][0]}')
print(f'{counts[0][1]}: {counts[1][1]}')

# Calculate the probabilities of each outcome
probability_0 = counts[1][0] / num_simulations
probability_1 = counts[1][1] / num_simulations

print("\nSimulation Results:")
print(f"Number of simulations: {num_simulations}")
print(f"Probability of outcome 0 (corresponding to |+>): {probability_0:.3f}")
print(f"Probability of outcome 1 (corresponding to |->): {probability_1:.3f}")

Question 4:

Describe the possible measurement outcome and give the probability for the outcome for this pair consisting of a state and a measurement basis:

\ket{\psi} = \frac{1}{\sqrt{2}}\left( \ket{0} - \ket{1} \right), {\ket{i}, \ket{-i}}

Solution

Again, since the state is measured in $∣i⟩,∣−i⟩{\ket{i}, \ket{-i}}$ , the possible measurement outcomes are obtaining the system in the state $∣i⟩\ket{i}$ or in the state $∣−i⟩\ket{-i}$ .

Now, there are two approaches to finding the probabilities of these outcomes. We will explore the fundamental (inner product) approach first.

Recall that:

Given a state $∣ψ⟩\ket{\psi}$ , the probability of each of the measurement outcomes in the given basis is the square of the inner product of the state with each of the measurement outcomes.

and (from the Bloch Sphere):

\begin{gather} \ket{i} = \frac{1}{\sqrt{2}}(\ket{0} + i\ket{1}) \ \bra{i} = \frac{1}{\sqrt{2}}(\bra{0} - i\bra{1}) \end{gather}

\begin{gather} \ket{-i} = \frac{1}{\sqrt{2}}(\ket{0} - i\ket{1}) \ \bra{-i} = \frac{1}{\sqrt{2}}(\bra{0} + i\bra{1}) \end{gather}

Therefore,

P∣i⟩=∣⟨i∣ψ⟩∣2\mathcal{P}{\ket{i}} = |\braket{i|\psi}|^2

P∣−i⟩=∣⟨−i∣ψ⟩∣2\mathcal{P}{\ket{-i}} = |\braket{-i|\psi}|^2

Let's expand $P∣i⟩\mathcal{P}_{\ket{i}}$ out first:

ParseError: KaTeX parse error: Expected '}', got '\right' at position 138: …{0} + i\ket{1})\̲r̲i̲g̲h̲t̲}|\left{\frac{1…

Given a complex number $z = a + bi$ , its magnitude $\sqrt{a^2 + b^2}$ . So, $\sqrt{1^2 + 1^2} = \sqrt{2}$ .

$∴\therefore$

\mathcal{P}_{\ket{i}}= \frac{1}{4}(\sqrt{2})^2 = \frac{1}{4}\cdot2 = \frac{1}{2}

In the same vein, $P∣−i⟩\mathcal{P}_{\ket{-i}}$ is estimated as follows:

ParseError: KaTeX parse error: Expected '}', got '\right' at position 140: …{0} - i\ket{1})\̲r̲i̲g̲h̲t̲}|\left{\frac{1…

Of course, they obey the Born rule.

A second approach is to flex some mathematical muscle and some intuition derived from the fact that the given state can be expressed in terms of the measurement basis.

Imagine $α\alpha$ and $β\beta$ coefficients such that:

\begin{equation} \tag{3i} \ket{\psi} = a\ket{i} + b\ket{-i} \end{equation}

Substituting $(1)$ and $(3)$ into $(3 i)$ :

\begin{align} \notag \frac{1}{\sqrt{2}}\left( \ket{0} - \ket{1} \right) &= \alpha \cdot \frac{1}{\sqrt{2}}\left( \ket{0} + i\ket{1} \right) + \beta \cdot \frac{1}{\sqrt{2}}\left( \ket{0} - i\ket{1} \right)\ \frac{1}{\sqrt{2}}\ket{0} + \frac{-1}{\sqrt{2}}\ket{1} &= \frac{\alpha + \beta}{\sqrt{2}}\ket{0} + \frac{\alpha i - \beta i}{\sqrt{2}}\ket{1} \end{align}

Comparing both sides of $(5)$ with the respective values of $∣0⟩\ket{0}$ and $∣1⟩\ket{1}$ :

\begin{align} \notag \frac{1}{\sqrt{2}}\ket{0} &= \frac{\alpha + \beta}{\sqrt{2}}\ket{0}\ \tag{3ii} \alpha + \beta &= 1 \end{align}

and

\begin{align} \notag \frac{-1}{\sqrt{2}}\ket{1} &= \frac{\alpha i - \beta i}{\sqrt{2}}\ket{1}\ \notag \alpha i - \beta i &= -1\ \notag i (\alpha - \beta) &= - 1\ \notag \alpha - \beta &= \frac{-1}{i} = \frac{-1}{i}\cdot\frac{i}{i} = \frac{-i}{i^2} = \frac{-i}{-1} \ \tag{3iii} \alpha - \beta &= i \end{align}

Solve $(3 ii)$ and $(3 iii)$ simultaneously by adding them together (substitution method),

ParseError: KaTeX parse error: Too many tab characters: & at position 77: …\alpha - \beta &̲= i \end{split}…

\begin{align} \notag 2\alpha &= 1 + i\ \tag{3iv} \alpha &= \frac{1+i}{2} \end{align}

Let's substitute $α\alpha$ in $(3 ii)$ :

\begin{align} \notag \frac{1+i}{2} + \beta &= 1\ \notag \beta &= 1 - \frac{1+i}{2}\ \tag{3v} \beta &= \frac{2 - (1+i)}{2} = \frac{1-i}{2} \end{align}

With these, our imagined expression becomes:

\ket{\psi} = \frac{1 + i}{2}\ket{i} + \frac{1 - i}{2} \ket{-i}

So,

\mathcal{P}_{\ket{i}} = \left|\frac{1}{2} (1 + i)\right|^{2} = \left|\frac{1}{2}\right|^{2} \times |(1 + i)|^2 = \frac{1}{4} \times \left(\sqrt{1^2 + (1)^2}\right)^2 = \frac{1}{4} \times 2 = \frac{1}{2}

\mathcal{P}_{\ket{-i}} = \left|\frac{1}{2} (1 - i)\right|^{2} = \left|\frac{1}{2}\right|^{2} \times |(1 - i)|^2 = \frac{1}{4} \times \left(\sqrt{1^2 + (-1)^2}\right)^2 = \frac{1}{4} \times 2 = \frac{1}{2}

Arriving at the same answers with the first approach, though longer.

Any of these approaches can be used to tackle problems of this nature. The first approach is quite a generalist!

Before closing up here, let's write a cirq simulation for this:

import cirq
import numpy as np

qubit = cirq.LineQubit(0)
psi_preparation = cirq.Circuit(cirq.H(qubit), cirq.Z(qubit))
print("Circuit for preparing psi:")
print(psi_preparation)

# This is based on some unitary transformations since, by default, `cirq.measure`
# perform standard measurements in the computational basis (Z-basis)
# and to measure in another basis, we need to do the unitary transformations that map
# this |i>, |-i> to the computational basis before performing the measurement
measurement_basis_transformation = cirq.Circuit(cirq.S(qubit)**-1, cirq.H(qubit))
measurement_circuit = psi_preparation + measurement_basis_transformation + cirq.Circuit(cirq.measure(qubit, key='result'))
print("\nCircuit for measurement in the {|i>, |-i>} basis:")
print(measurement_circuit)

simulator = cirq.Simulator()
num_repetitions = 1000
results = simulator.run(measurement_circuit, repetitions=num_repetitions)
measurement_counts = results.histogram(key='result')
print("\nMeasurement counts:", measurement_counts)

probability_i = measurement_counts.get(0, 0) / num_repetitions
probability_minus_i = measurement_counts.get(1, 0) / num_repetitions
print(f"Probability of outcome corresponding to |i>: {probability_i:.3f}")
print(f"Probability of outcome corresponding to |-i>: {probability_minus_i:.3f}")

Outro

References
[@Rieffel2011]: Rieffel, E. G., & Polak, W. H. (2014). Quantum Computing: A Gentle Introduction. MIT Press. https://books.google.com/books?id=CQ3YoAEACAAJ
[@Bernhardt2019]: Bernhardt, C. (2019). Quantum Computing for Everyone. The MIT Press.
[@Schumacher1995]: Schumacher, B. (1995). Quantum coding. Physical Review A, 51(4), 2738-2747.
[@Swayne2024]: Swayne, M. (2024). Chinese scientists report using quantum computer to hack military-grade encryption. The Quantum Insider. https://thequantuminsider.com/2024/10/11/chinese-scientists-report-using-quantum-computer-to-hack-military-grade-encryption/. Retrieved April 17, 2025
[@Lopez2025]: Lopez, S., Hudek, T., Li, M., Hansen, D. P., Sánchez, E. G. & Gronlund, C. J. (2025). Dirac notation in quantum computing. Microsoft Learn. https://learn.microsoft.com/en-us/azure/quantum/concepts-dirac-notation. Retrieved April 17, 2025
[@Shor2025]: Shor, P. (2022). Lecture 2: Quantum Computation. MIT Mathematics. https://math.mit.edu/~shor/435-LN/Lecture_02.pdf. Retrieved April 17, 2025
[@Born1926]: Born, M. (1926). Zur Quantenmechanik der Stoßvorgänge. Zeitschrift für Physik, 37(12), 863-867.

How to Build a Dual-LLM Chat Application with Next.js, Python, and WebSocket Streaming

John Owolabi Idogun — Thu, 27 Feb 2025 04:00:48 +0000

Introduction

In the age of rapid AI advancement, software engineers are frequently tasked with building sophisticated chatbots. This article provides a practical guide to developing a real-time, response-streaming chatbot from scratch using transformer models, autoencoding (BERT family), and autoregressive (GPT family) techniques. We'll learn how to build a chatbot capable of contextual conversations, processing user text and image/PDF uploads for enhanced understanding. We'll utilize Python (aiohttp) and React (Next.js) to deliver a seamless, interactive experience. Let's dive in!

Prerequisite

No prior expertise is strictly necessary, but some foundation in AI, software engineering, Python coding, and frontend development with a framework like React, Svelte, Vue, or Angular will be beneficial. If you're new to aiohttp or Svelte, we highly suggest exploring this previous article series as a helpful primer. Don't worry if you're not an expert (I ain't).

To set up your project, first create a root folder genai-chatbot and then navigate into the backend subdirectory:

mkdir genai-chatbot && cd genai-chatbot && mkdir backend && cd backend

In the backend directory, create a Python virtual environment and install dependencies from requirements.txt:

python3 -m venv virtualenv && source virtualenv/bin/activate && pip install -r requirements.txt

Ensure you create a requirements.txt file in the backend directory with the following content:

aiohttp
aiodns 
transformers 
torch 
torch-vision
pymupdf
pytesseract

For the frontend, navigate to the root directory (genai-chatbot) and use create-next-app to generate a React frontend with TypeScript:

cd ..
npx create-next-app@latest react-frontend --typescript --eslint --app

Follow the prompts from create-next-app.

Tailwind CSS Setup Recommendation:

Please note that create-next-app might include an older version of Tailwind CSS by default. For the best experience, avoiding the default Tailwind CSS setup during the create-next-app configuration is recommended. Instead, follow the official Tailwind CSS Next.js installation guide to install the latest version, ensuring you install Tailwind CSS and related packages as devDependencies.

Source code

Sirneij / genai-chatbot

A streaming chatbot powered by Large Language Models

GenAI ChatBot

A full-stack chat application featuring autoregressive and masked language models with streaming responses and context-aware interactions.

Overview

Frontend: Next.js-based chat interface with real-time WebSocket streaming
Backend: Python (aiohttp) server handling LLMs and file processing

Quick Start

Clone the repository and enter the directory:

git clone https://github.com/Sirneij/genai-chatbot.git
cd genai-chatbot

Start the backend server (in a new terminal):

cd backend
python -m venv virtualenv
source virtualenv/bin/activate  # On Windows, use: virtualenv\Scripts\activate
pip install -r requirements.txt
adev runserver main.py  # For development with auto-reload
# Or use: python main.py  # For production

Start the frontend app (in another terminal):

cd react-frontend
npm install
npm run dev

The application will be available at http://localhost:3000

Features

Real-time streaming responses
Support for both autoregressive and masked LLMs
File upload support (PDFs/Images)
Rich text rendering:
- Markdown support
- Code syntax highlighting
- KaTeX math expressions
- Unicode emoji support
Dark mode and responsive design

Project Structure

…

View on GitHub

Implementation

Step 1: Theory - Autoregressive vs Autoencoding models

Two primary approaches emerge within the transformer model architecture: autoregressive and autoencoding. Autoregressive models, like those using the decoder component of a transformer, predict the next word (token) in a sequence based solely on the preceding words (a strictly left-to-right approach). This sequential generation method is ideal for tasks like text generation and response streaming. In contrast, autoencoding models (also known as masked language models) leverage the encoder part of the transformer. They operate bidirectionally, considering both preceding and subsequent words in a sentence to understand the context and produce representations. Refer to Hugging Face's model summary for a detailed comparison. Pretraining methodology is a key differentiator: autoregressive models are pretrained to predict the next token, making them naturally suited for text generation, while autoencoding models excel at contextual understanding, making them strong for tasks like question answering based on documents.

In this project, we will practically explore the distinct text generation capabilities of both model types in real-time. We will use microsoft/Phi-3-mini-4k-instruct (autoregressive) and deepset/roberta-base-squad2 (autoencoding) for demonstration, acknowledging resource limitations and focusing on showcasing the core differences.

Step 2: Python asynchronous backend with aiohttp

We'll start with the project structure from Building and Rigorously Testing a WebSocket and HTTP Server. Populate src/app/__init__.py with the following code, noting the key adaptations for our chatbot:

import json
from weakref import WeakSet

from aiohttp import WSCloseCode, web
from aiohttp.web import Request, Response, WebSocketResponse

from src.utils.auto_chat_engine import (
    cleanup_auto_model,
    gpt_question_and_answer,
    prepare_auto_tokenizer_and_model,
)
from src.utils.chat_engine import (
    prepare_qa_tokenizer_and_model,
    squad_question_answering,
)
from src.utils.extract import extract_text_from_file
from src.utils.settings import base_settings

WEBSOCKETS = web.AppKey('websockets', WeakSet[WebSocketResponse])


async def start_background_tasks(app: web.Application) -> None:
    """Initialize application background tasks."""
    app[WEBSOCKETS] = WeakSet()
    await prepare_auto_tokenizer_and_model(base_settings.MODEL_NAME)
    await prepare_qa_tokenizer_and_model(base_settings.QA_MODEL_NAME)


async def cleanup_app(app: web.Application) -> None:
    """Cleanup WebSocket connections on shutdown."""
    # Cleanup models
    await cleanup_auto_model()
    # Close all WebSocket connections
    for websocket in set(app[WEBSOCKETS]):  # type: ignore
        await websocket.close(code=WSCloseCode.GOING_AWAY, message=b'Server shutdown')


async def extract_text(request: Request) -> Response:
    """Extract text from PDF and image files."""
    data = await request.post()
    files = data.getall('file')
    if not files:
        return web.json_response({'error': 'No files provided'}, status=400)

    extracted_text = []
    for file in files:
        if file.content_type not in ['application/pdf', 'image/jpeg', 'image/png']:
            return web.json_response({'error': 'Invalid file type'}, status=400)

        text = await extract_text_from_file(file.file, file.content_type)
        extracted_text.append(text)

    base_settings.context = '\n'.join(extracted_text)

    return web.json_response({'success': 'Text extracted successfully'})


async def chat_handler(request: Request) -> Response:
    """Handle WebSocket connections."""
    ws = WebSocketResponse()
    await ws.prepare(request)

    request.app[WEBSOCKETS].add(ws)

    async for msg in ws:
        if msg.type == web.WSMsgType.TEXT:
            try:
                data = json.loads(msg.data)
                question_type = data.get('type')
                question = data.get('question', '').strip()
                if not question:
                    await ws.send_str('Error: No question provided.')
                    continue

                if question_type == 'auto':
                    # Stream response token by token.
                    async for token in gpt_question_and_answer(question):
                        await ws.send_json({'answer': token})
                elif question_type == 'masked':
                    # Use squad question answering (non-streamed).
                    answer = await squad_question_answering(question)
                    await ws.send_json({'answer': answer})
                else:
                    await ws.send_str('Error: Unknown question type.')
            except Exception as e:
                await ws.send_str(f'Error processing message: {str(e)}')
        elif msg.type == web.WSMsgType.ERROR:
            request.app[WEBSOCKETS].remove(ws)
            break

    request.app[WEBSOCKETS].remove(ws)

    return ws


def init_app() -> web.Application:
    """Initialize the application."""
    app = web.Application()

    # Add routes
    app.router.add_post('/api/extract', extract_text)
    app.router.add_get('/chat', chat_handler)

    # Add startup/cleanup handlers
    app.on_startup.append(start_background_tasks)
    app.on_shutdown.append(cleanup_app)

    return app

This __init__.py file builds upon the structure from the recommended series. The key differences for our chatbot are:

start_background_tasks (on startup) loads AI models, and cleanup_app (on shutdown) unloads them, improving resource handling.
The extract_text handler accepts and processes multiple uploaded files to create a combined context.
We use direct aiohttp WebSocket APIs in chat_handler. We support auto and masked types representing the use of autogenerative and masked language models respectively.

Next, let's delve into the files within the src/utils directory, starting with settings.py.

import logging

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    datefmt='%Y-%m-%d %H:%M:%S',
)

logger = logging.getLogger(__name__)


class Settings:
    logger = logger
    SYSTEM_PROMPT = 'You are an expert Q&A assistant. Provide accurate answers based on the context.'
    context = None
    MODEL_NAME = 'microsoft/Phi-3-mini-4k-instruct'
    QA_MODEL_NAME = 'deepset/roberta-base-squad2'


base_settings = Settings()

The settings.py file primarily houses our logger configuration and defines several important constants used in the application, including the system prompt, the default language model (MODEL_NAME), and the question answering model (QA_MODEL_NAME).

Let's examine base.py next:

import torch


def get_device() -> tuple[torch.device, str]:
    if torch.cuda.is_available():
        return torch.device('cuda'), 'CUDA (NVIDIA GPU)'
    elif torch.backends.mps.is_available():
        return torch.device('mps'), 'MPS (Apple Metal)'
    else:
        return torch.device('cpu'), 'CPU'


async def get_stopping_strings(type: str, question: str, context: str | None = None) -> tuple[str, list[str]]:
    """Get stopping strings based on the question type."""
    ctx = f'\nUse this as the base Context: {context}\n' if context else ''
    STOPPING_STRINGS = {
        'exNormal': {
            'prompt': f'Answer the question, succinctly, using markdown formatting and katex for math.{ctx}. Do not use unicodes, if you must, use \\ such as \\u1F60A for smile.\nAfter providing your complete answer, you must conclude your response with \n§\n as the final line, with no text following it. \n\nQuestion: {question}\n\nAnswer:',
            'end': ['§', '\n§', '§\n', ' §', '\n§\n'],
        },
    }

    return STOPPING_STRINGS[type]['prompt'], STOPPING_STRINGS[type]['end']

get_device intelligently detects the available hardware and returns the appropriate PyTorch device (CUDA GPU, Apple Metal (MPS), or CPU) along with a descriptive string. get_stopping_strings plays a crucial role in prompt construction for the autoregressive model. It dynamically generates a prompt and a list of stopping strings based on the question type and optional context. The prompt instructs the model to provide a concise answer using Markdown formatting and KaTeX for mathematical expressions. A key challenge with autoregressive models is their tendency to generate repetitive text. To mitigate this, the prompt explicitly instructs the model to conclude its response with the "§" symbol, an uncommon character chosen to signal the end of the answer. The application then programmatically halts text generation upon encountering this symbol. While using the model's native end-of-sequence token would be ideal, I had no luck with it.

The auto_chat_engine.py file is the central component responsible for generating text using the configured language model. It orchestrates the model's loading, manages the text generation process, and handles device-specific optimizations.

import asyncio

import torch
from transformers import AutoModelForCausalLM, AutoTokenizer, PreTrainedTokenizerFast

from src.utils.base import get_device, get_stopping_strings
from src.utils.settings import base_settings

TOKENIZER, MODEL = None, None


async def prepare_auto_tokenizer_and_model(model_name: str) -> None:
    """
    Prepare and load the tokenizer and model with hardware-specific optimizations.

    Args:
        model_name (str): The name of the model to load.

    Returns:
        tuple: The loaded tokenizer and model.
    """

    global TOKENIZER, MODEL

    if TOKENIZER is not None and MODEL is not None:
        return

    device, _ = get_device()

    # Load tokenizer
    tokenizer = AutoTokenizer.from_pretrained(model_name, padding_side='left', truncation_side='left', use_fast=True)
    tokenizer.pad_token = tokenizer.eos_token

    TOKENIZER = tokenizer

    # Model-specific settings
    load_kwargs = {}
    if 'phi-3' in model_name.lower():
        load_kwargs['trust_remote_code'] = True
        load_kwargs['attn_implementation'] = 'eager'  # No CUDA, so use eager mode
    elif 'starcoder' in model_name.lower():
        load_kwargs['trust_remote_code'] = True

    # Load model based on device
    if device.type == 'cpu':
        # Use float32 for CPU and apply dynamic quantization
        model = AutoModelForCausalLM.from_pretrained(
            model_name,
            torch_dtype=torch.float32,
            **load_kwargs,
        )
        model = torch.quantization.quantize_dynamic(model, {torch.nn.Linear}, dtype=torch.qint8)
    elif device.type == 'mps':
        # Use bfloat16 for MPS
        model = AutoModelForCausalLM.from_pretrained(
            model_name,
            torch_dtype=torch.bfloat16,
            **load_kwargs,
        ).to(device)

    model.eval()

    MODEL = model


async def cleanup_auto_model() -> None:
    """Clear model from memory, important for Apple Silicon"""
    global MODEL, TOKENIZER
    if torch.backends.mps.is_available():
        torch.mps.empty_cache()
    del MODEL
    del TOKENIZER


def top_k_top_p_filtering(logits, top_k=0, top_p=1.0, filter_value=-float("Inf")):
    """
    Filter a distribution of logits using top-k and nucleus (top-p) filtering.
    """
    if top_k > 0:
        values, _ = torch.topk(logits, top_k)
        kth_value = values[..., -1, None]
        logits = torch.where(logits < kth_value, torch.full_like(logits, filter_value), logits)

    if top_p < 1.0:
        sorted_logits, sorted_indices = torch.sort(logits, descending=True)
        cumulative_probs = torch.cumsum(torch.softmax(sorted_logits, dim=-1), dim=-1)
        sorted_indices_to_remove = cumulative_probs > top_p
        sorted_indices_to_remove[..., 1:] = sorted_indices_to_remove[..., :-1].clone()
        sorted_indices_to_remove[..., 0] = 0
        indices_to_remove = sorted_indices[sorted_indices_to_remove]
        logits[indices_to_remove] = filter_value
    return logits


async def stream_chat_response(
    prompt: str,
    tokenizer: PreTrainedTokenizerFast,
    model,
    stopping_strings: list[str],
    max_new_tokens: int = 100,
    temperature: float = 0.5,
    top_k: int = 50,
    top_p: float = 0.9,
    repetition_penalty: float = 1.5,
    repetition_window: int = 10,
):
    device = next(model.parameters()).device
    eos_token_id = tokenizer.eos_token_id
    inputs = tokenizer(prompt, return_tensors='pt')
    input_ids = inputs['input_ids'].to(device)
    attention_mask = inputs['attention_mask'].to(device)
    full_generated_ids = input_ids
    previous_generated_text = ''
    generated_tokens = []

    for _ in range(max_new_tokens):
        with torch.no_grad():
            if device.type == 'cuda':
                with torch.autocast(device_type='cuda', dtype=torch.float16):
                    outputs = model(full_generated_ids, attention_mask=attention_mask)
            elif device.type == 'mps':
                with torch.autocast(device_type='mps', dtype=torch.bfloat16):
                    outputs = model(full_generated_ids, attention_mask=attention_mask)
            else:  # CPU
                outputs = model(full_generated_ids, attention_mask=attention_mask)

        next_token_logits = outputs.logits[:, -1, :]
        # Apply repetition penalty
        if repetition_penalty != 1.0:
            for token in set(generated_tokens[-20:]):
                next_token_logits[:, token] /= repetition_penalty

        scaled_logits = next_token_logits / temperature
        filtered_logits = top_k_top_p_filtering(scaled_logits[0], top_k=top_k, top_p=top_p)
        probabilities = torch.softmax(filtered_logits, dim=-1)
        next_token = torch.multinomial(probabilities, num_samples=1).unsqueeze(0)
        current_token = next_token.item()

        full_generated_ids = torch.cat([full_generated_ids, next_token], dim=-1)
        attention_mask = torch.cat([attention_mask, torch.ones_like(next_token)], dim=-1)
        generated_tokens.append(current_token)

        if current_token == eos_token_id:
            break

        current_generated_text = tokenizer.decode(
            full_generated_ids[0][input_ids.shape[1] :].cpu(), skip_special_tokens=True
        )

        # Check stopping conditions
        if any(stop_str in current_generated_text for stop_str in stopping_strings):
            break
        if (
            len(generated_tokens) >= 2 * repetition_window
            and generated_tokens[-repetition_window:] == generated_tokens[-2 * repetition_window : -repetition_window]
        ):
            break

        diff = current_generated_text[len(previous_generated_text) :]
        if diff:
            yield diff
            previous_generated_text = current_generated_text

        if device.type == 'mps':
            await asyncio.sleep(0.001)  # Yield event loop for MPS
        elif device.type == 'cpu':
            await asyncio.sleep(0.005)  # Reduce CPU load
        else:
            await asyncio.sleep(0)

    # Return the final text
    yield '[END]'


async def gpt_question_and_answer(question: str):
    """Stream an answer with repetition penalty and better stopping checks."""

    prompt, stopping_strings = await get_stopping_strings('exNormal', question, context=base_settings.context)
    async for chunk in stream_chat_response(
        prompt,
        TOKENIZER,
        MODEL,
        stopping_strings=stopping_strings,
        max_new_tokens=2000,  # Increase this for longer outputs
        top_p=0.95,
    ):
        yield chunk

prepare_auto_tokenizer_and_model initializes the tokenizer and model. To avoid redundant or repetitive loading, we used global variables and adapted the loading process based on the available device (CPU, MPS, CUDA).

torch.quantization.quantize_dynamic is a nifty way to quantize the model so that it will run faster and require lower memory. If your machine supports cuda, you will need to add your block in the code.

cleanup_auto_model primarily releases the model and tokenizer from memory. This is particularly important on memory-constrained devices like those with Apple Silicon (MPS), where memory leaks can quickly degrade performance.

top_k_top_p_filtering implements two common techniques for controlling the diversity and quality of generated text. Top-k filtering limits the model's choices to the k most likely tokens, while top-p (nucleus) sampling considers the smallest set of tokens whose cumulative probability exceeds p. In simple terms, they help to strike a balance between generating creative and coherent text by preventing the model from generating nonsensical or irrelevant tokens.

The stream_chat_response function is the heart of our real-time chatbot's text generation. It takes a prompt, the model, and configurations to generate streaming text output. For performance, within the generation loop, we:

Optimize Device Usage: Move model parameters to the active device for faster access.
Disable Gradients: Use torch.no_grad() to skip unnecessary gradient calculations during inference.
Enable Auto-Precision: Utilize torch.autocast for faster, mixed-precision operations.

The generation process automatically stops when it encounters our stopping character (code lines 151-152). To ensure smooth streaming, the function also briefly releases memory between tokens. Finally, after generating the complete response, stream_chat_response yields [END] to signal to the frontend that the stream is complete.

Disclaimer: Learning in Progress

As a learner in this field, I welcome any corrections or suggestions regarding my approach. Please feel free to point out any errors, and let's learn together!

gpt_question_and_answer serves as a high-level interface for generating answers to questions. It retrieves the appropriate prompt and stopping strings using get_stopping_strings and then calls stream_chat_response to generate the text. You can modify the parameters here to your taste.

Finally, let's examine the chat_engine.py file. This module leverages pre-trained BERT-based models for extracting answers from a given context.

import asyncio

from transformers import pipeline

from src.utils.base import get_device
from src.utils.settings import base_settings

QA_PIPELINE = None


async def prepare_qa_tokenizer_and_model(model_name: str) -> None:
    """
    Prepare and load pretrained QA pipeline with hardware-specific optimizations.
    """
    global QA_PIPELINE
    if QA_PIPELINE is None:
        QA_PIPELINE = pipeline(
            'question-answering',
            model=model_name,
            tokenizer=model_name,
            device=-1 if get_device()[0].type == 'cpu' else 0,
        )
        base_settings.logger.info('QA pipeline initialized')


async def squad_question_answering(question: str) -> str:
    """Optimized QA"""
    if not base_settings.context:
        return {'error': 'Please upload a document first.'}

    if QA_PIPELINE is None:
        return {'error': 'QA system not initialized'}

    base_settings.logger.info(f'Question: {question}')
    base_settings.logger.info(f'Context: {base_settings.context}')

    try:
        result = await asyncio.to_thread(
            QA_PIPELINE,
            question=question,
            context=base_settings.context,
        )
        return result['answer']

    except Exception as e:
        base_settings.logger.error(f'QA Error: {str(e)}')
        return {'error': 'An error occurred during QA'}

prepare_qa_tokenizer_and_model initializes the question-answering pipeline. The primary motivation behind using a global QA_PIPELINE, as previously stated, is to avoid repeatedly loading the model, which is a resource-intensive operation. In squad_question_answering, a key design decision here is the enforcement of a context. BERT-based models, while powerful, generally perform best when given a specific context to ground their answers. To prevent blocking the main thread, the actual question-answering process is offloaded to a separate asynchronous thread using asyncio.to_thread.

Step 3: Building a React Frontend with Next.js and Tailwind CSS

While our server backend is functional and can be interacted with via command-line tools like wscat (using a command such as wscat -c ws://localhost:PORT/chat followed by a prompt like {"type": "auto", "question": "Write Python code for Fibonacci sequence"}), this approach is not user-friendly. To make our AI chatbot accessible to a wider audience, including those unfamiliar with the terminal, we need a graphical user interface. We'll leverage Next.js, a React framework known for its performance and developer experience, along with Tailwind CSS for styling.

Our frontend application consists of a single page with a few key components. Let's start by examining react-frontend/src/app/layout.tsx:

import type { Metadata } from "next";
import { Geist, Geist_Mono } from "next/font/google";
import "katex/dist/katex.min.css";
import "./globals.css";
import ThemeSwitcher from "$/app/ui/layout/ThemeSwitcher";

const geistSans = Geist({
  variable: "--font-geist-sans",
  subsets: ["latin"],
});

const geistMono = Geist_Mono({
  variable: "--font-geist-mono",
  subsets: ["latin"],
});

export const metadata: Metadata = {
  title: "AI Chatbot with Next.js and Python | John Owolabi Idogun",
  description:
    "Build an AI chatbot with Next.js and Python using WebSockets by John Owolabi Idogun.",
};

export default function RootLayout({
  children,
}: Readonly<{
  children: React.ReactNode;
}>) {
  return (
    <html lang="en">
      <body
        className={`${geistSans.variable} ${geistMono.variable} antialiased font-[family-name:var(--font-geist-sans)] bg-[#ffffff] text-[#171717] dark:bg-[#0a0a0a] dark:text-[#ededed] min-h-screen`}
      >
        <ThemeSwitcher />
        <main className="flex flex-col min-h-screen mx-auto max-w-10/12">
          <div className="flex-1 w-full">{children}</div>
        </main>
      </body>
    </html>
  );
}

This layout.tsx file defines the root layout of our Next.js application. It's largely based on the default code generated by create-next-app and enhanced with Tailwind CSS integration. We've also integrated KaTeX for rendering mathematical expressions (remember to install the katex package and its TypeScript definitions, as well as marked.js for Markdown processing). The ThemeSwitcher component, responsible for toggling between light and dark themes, is also included.

Here's the code for the ThemeSwitcher component:

"use client";

import { useState, useEffect } from "react";
import { MoonIcon, SunIcon } from "$/app/ui/icons/base";

export default function ThemeSwitcher() {
  const [theme, setTheme] = useState("light");

  useEffect(() => {
    const savedTheme = localStorage.getItem("theme") || "light";
    setTheme(savedTheme);
    document.documentElement.classList.toggle("dark", savedTheme === "dark");
  }, []);

  const toggleTheme = () => {
    const newTheme = theme === "light" ? "dark" : "light";
    setTheme(newTheme);
    localStorage.setItem("theme", newTheme);
    document.documentElement.classList.toggle("dark");
  };

  return (
    <button
      onClick={toggleTheme}
      className="fixed top-4 right-4 p-2 rounded-full bg-gray-200 dark:bg-gray-800 hover:bg-gray-300 dark:hover:bg-gray-700 transition-colors"
      aria-label="Toggle theme"
    >
      {theme === "light" ? (
        <MoonIcon className="h-5 w-5" />
      ) : (
        <SunIcon className="h-5 w-5" />
      )}
    </button>
  );
}

For Svelte developers, React's useEffect hook serves a similar purpose to Svelte 5's $effect block. Both allow you to run code in response to changes (in useEffect you need to list the dependencies, you don't need that in Svelte), and critically, both provide a safe context for interacting with HTML elements.

Now is the turn is react-frontend/src/app/page.tsx:

"use client";

import { useCallback, useState } from "react";
import ChatContainer from "$/app/ui/chat/ChatContainer";
import ChatInput from "$/app/ui/chat/ChatInput";
import { Logo } from "$/app/ui/icons/base";
import { useWebSocket } from "$/app/lib/hooks/useWebSocket";
import { Message } from "$/app/lib/types";
import { useAutoScroll } from "$/app/lib/hooks/useAutoScroll";
import { BASE_WS_URL } from "./lib/constants";

export default function Home() {
  const [messages, setMessages] = useState<Message[]>([]);
  const scrollRef = useAutoScroll<HTMLDivElement>(messages);

  const handleSend = (message: string, type: string) => {
    const newMessage = {
      id: Date.now(),
      text: message,
      sender: "user" as const,
    };

    // Add a loading message for the bot
    const loadingMessage = {
      id: Date.now() + 1,
      text: "",
      sender: "bot" as const,
      loading: true,
      complete: false,
    };

    setMessages((prev) => [...prev, newMessage, loadingMessage]);
    sendMessage(message, type);
  };

  const handleBotMessage = useCallback(
    (content: string, isComplete?: boolean) => {
      setMessages((prev) => {
        const lastMessage = prev[prev.length - 1];

        if (isComplete && lastMessage?.sender === "bot") {
          const updatedMessages = [...prev];
          updatedMessages[prev.length - 1] = {
            ...lastMessage,
            complete: true,
            loading: false,
          };
          return updatedMessages;
        }

        if (!content) return prev;

        if (
          !lastMessage ||
          lastMessage.sender !== "bot" ||
          lastMessage.complete
        ) {
          return [
            ...prev,
            {
              id: Date.now(),
              text: content,
              sender: "bot",
              complete: false,
              loading: false,
            },
          ];
        }

        const updatedMessages = [...prev];
        updatedMessages[prev.length - 1] = {
          ...lastMessage,
          text: lastMessage.text + content,
          loading: false,
        };

        return updatedMessages;
      });
    },
    []
  );

  const { sendMessage, isConnected, error } = useWebSocket(
    `${BASE_WS_URL}/chat`,
    handleBotMessage
  );

  return (
    <div className="flex flex-col flex-1 h-[calc(100vh-10rem)]">
      {!messages.length ? (
        <div className="flex flex-col items-center justify-center flex-1">
          <div className="flex flex-col items-center max-w-3xl w-full mx-auto">
            <Logo className="h-16 w-16 mb-6 text-[#171717] dark:text-[#ededed]" />
            <p className="text-gray-600 dark:text-gray-400 mb-8">
              Ask me anything! I'm here to help.
            </p>
            <div className="w-full">
              <ChatInput onSend={handleSend} messageCount={messages.length} />
            </div>
          </div>
        </div>
      ) : (
        <>
          <div ref={scrollRef} className="flex-1 overflow-y-auto no-scrollbar">
            <ChatContainer messages={messages} />
          </div>
          <ChatInput onSend={handleSend} messageCount={messages.length} />
        </>
      )}
    </div>
  );
}

This file represents the main page component of our chat application, housing the core logic for message handling and UI rendering. The Home component manages the chat interface, including displaying messages, handling user input, and communicating with the backend. The handleSend function adds a user message and a loading message to the state, while handleBotMessage updates the UI with the bot's responses as they stream in.

The custom useWebSocket hook encapsulates all the WebSocket logic, including connection management, message handling, and error handling:

import { useState, useEffect, useCallback } from "react";

type WebSocketMessage = {
  type: string;
  question: string;
};

export const useWebSocket = (
  url: string,
  onMessage: (content: string, isComplete?: boolean) => void
) => {
  const [socket, setSocket] = useState<WebSocket | null>(null);
  const [isConnected, setIsConnected] = useState(false);
  const [error, setError] = useState<string | null>(null);

  useEffect(() => {
    const ws = new WebSocket(url);

    ws.onopen = () => {
      console.log("WebSocket Connected");
      setIsConnected(true);
      setError(null);
    };

    ws.onmessage = (event) => {
      try {
        const data = JSON.parse(event.data);
        // Handle the streaming response
        if (data.answer === "[END]") {
          onMessage("", true); // Signal completion
        } else {
          onMessage(data.answer, false);
        }
      } catch (e) {
        console.error("Error parsing message:", e);
      }
    };

    ws.onclose = () => {
      console.log("WebSocket Disconnected");
      setIsConnected(false);
      setTimeout(() => setSocket(new WebSocket(url)), 3000);
    };

    ws.onerror = (event) => {
      console.error("WebSocket error:", event);
      setError("WebSocket error occurred");
    };

    setSocket(ws);

    return () => {
      ws.close();
    };
  }, [url, onMessage]);

  const sendMessage = useCallback(
    (message: string, type: string) => {
      if (socket && isConnected) {
        const payload: WebSocketMessage = {
          type: type,
          question: message,
        };
        socket.send(JSON.stringify(payload));
      } else {
        console.log("Socket not ready:", {
          isConnected,
          socketExists: !!socket,
        });
      }
    },
    [socket, isConnected]
  );

  return { sendMessage, isConnected, error };
};

handleSend function adds a temporary "loading" message to the chat interface. This is a simple but effective UI trick to provide immediate feedback to the user while waiting for the AI bot's response. Without this, the user might perceive a delay or lack of responsiveness which is unwanted. handleBotMessage function is designed to handle the streaming responses from the AI bot. It efficiently updates the chat interface by appending new content to the last bot message. The isComplete flag, triggered by the "[END]" signal from the backend, ensures that the loading indicator is removed when the bot finishes generating its response. Why useCallback in handleBotMessage? Its primary reason is to prevent unnecessary re-renders of components that depend on these functions and it does that by memoizing them so that they only change when their dependencies change.

To wrap up, let's look at the react-frontend/src/app/ui/chat/ChatContainer.tsx and react-frontend/src/app/ui/chat/ChatInput.tsx.

import React from "react";
import ChatMessage from "$/app/ui/chat/ChatMessage";
import { Message } from "$/app/lib/types";

interface ChatContainerProps {
  messages: Message[];
}

const ChatContainer: React.FC<ChatContainerProps> = ({ messages }) => {
  return (
    <div className="flex flex-col h-full overflow-y-auto p-4">
      <div className="flex-1 overflow-y-scroll">
        {messages.map((message) => (
          <ChatMessage key={message.id} message={message} />
        ))}
      </div>
    </div>
  );
};

export default ChatContainer;

The ChatContainer component renders the list of ChatMessage components. It's a straightforward component that iterates over the messages and displays each one.

import { marked } from "marked";
import { Message } from "$/app/lib/types";
import { ThinkingAnimation } from "$/app/ui/reusables";
import "katex/dist/katex.min.css";
import katex from "katex";
import hljs from "highlight.js";
import "highlight.js/styles/base16/horizon-dark.min.css";

// Helper function to preprocess Unicode characters
const preprocessUnicode = (text: string): string => {
  // Handle escaped Unicode sequences
  const unescaped = text.replace(
    /\\u\{?([a-fA-F0-9]{4,6})\}?/g,
    (_, codePoint) => String.fromCodePoint(parseInt(codePoint, 16))
  );

  return unescaped;
};

marked.setOptions({
  gfm: true,
  breaks: true,
  pedantic: false,
});

// Custom renderer to handle streaming better
const renderer = new marked.Renderer();

// Add math support
const mathRenderer = {
  name: "math",
  level: "inline",
  start(src: string) {
    return src.match(/\$/)?.index;
  },
  tokenizer(src: string) {
    const match = src.match(/^\$\$([^$\n]+?)\$\$|^\$([^$\n]+?)\$/);
    if (match) {
      const isDisplay = match[0].startsWith("$$");
      return {
        type: "math",
        raw: match[0],
        text: (isDisplay ? match[1] : match[2]).trim(),
        isDisplay,
      };
    }
  },
  renderer(token: any) {
    try {
      return katex.renderToString(token.text, {
        throwOnError: false,
        displayMode: token.isDisplay,
      });
    } catch (err) {
      return token.raw;
    }
  },
};

const codeHighlightExtension = {
  name: "code-highlight",
  level: "block",
  start(src: string) {
    return src.match(/^```
{% endraw %}
/)?.index;
  },
  tokenizer(src: string) {
    const match = src.match(/^
{% raw %}
```(\S*)\n([\s\S]*?)```
{% endraw %}
/);
    if (match) {
      return {
        type: "code-highlight",
        raw: match[0],
        lang: match[1],
        text: match[2].trim(),
      };
    }
  },
  renderer(token: any) {
    if (token.lang && hljs.getLanguage(token.lang)) {
      try {
        const highlighted = hljs.highlight(token.text, {
          language: token.lang,
          ignoreIllegals: true,
        }).value;
        return {% raw %}`<pre><code class="hljs language-${token.lang}">${highlighted}</code></pre>`{% endraw %};
      } catch (err) {
        console.error("Highlight.js error:", err);
      }
    }
    return {% raw %}`<pre><code>${token.text}</code></pre>`{% endraw %};
  },
};

marked.use({
  extensions: [mathRenderer, codeHighlightExtension],
});

interface ChatMessageProps {
  message: Message;
}

export default function ChatMessage({ message }: ChatMessageProps) {
  return (
    <div
      className={{% raw %}`flex ${
        message.sender === "user" ? "justify-end" : "justify-start"
      } mb-4`{% endraw %}}
    >
      <div
        className={{% raw %}`max-w-[80%] p-3 rounded-lg ${
          message.sender === "user"
            ? "bg-[#f5f5f5] text-[#171717] dark:bg-[#1a1a1a] dark:text-[#ededed]"
            : "bg-[#fafafa] text-[#171717] dark:bg-[#141414] dark:text-[#ededed]"
        }`{% endraw %}}
      >
        {message.loading ? (
          <ThinkingAnimation />
        ) : (
          <div className="prose article-content max-w-none animate-fade-in">
            {message.sender === "bot" ? (
              <span
                dangerouslySetInnerHTML={{
                  __html: marked(preprocessUnicode(message.text), { renderer }),
                }}
              />
            ) : (
              <span
                dangerouslySetInnerHTML={{
                  __html: marked(preprocessUnicode(message.text), { renderer }),
                }}
              />
            )}
          </div>
        )}
      </div>
    </div>
  );
}
{% raw %}

The ChatMessage component is responsible for rendering individual chat messages. The key feature of this component is its use of marked.js to render Markdown content, with extensions for syntax highlighting (using highlight.js) and math rendering (using KaTeX). The component also includes logic to display a "thinking" animation while the bot is generating a response.

Lastly, let's look at react-frontend/src/app/ui/chat/ChatInput.tsx.

It provides the user interface for entering and sending messages. It includes a text input, controls for toggling "Masked" and "Auto" modes, and a file upload button (you must select the masked mode to upload). The component was inspired by X's Grok 3, 😉.

An important to note is that in handleFileUpload, I used a NextJS action (uploadFiles) to communicate with the server for files upload:


typescript
"use server";

import { BASE_API_URL } from "$/app/lib/constants";

export async function uploadFiles(formData: FormData) {
  try {
    const response = await fetch(`${BASE_API_URL}/api/extract`, {
      method: "POST",
      body: formData,
    });

    if (!response.ok) {
      throw new Error("Upload failed");
    }

    return await response.json();
  } catch (error) {
    console.error("Upload error:", error);
    throw error;
  }
}

The key reason for using a server action is to avoid CORS errors such as Cross-Origin Request Blocked: The Same Origin Policy disallows reading the remote resource at .... By executing the upload logic on the server, we bypass the browser's same-origin policy and can communicate with the backend API without requiring CORS configuration.

I am sorry for the long article. I didn't want to split so I can be forced to just finish it at a go. There are still a couple of cleanups to do but I will leave those to you. Bye for now.

Outro

Enjoyed this article? I'm a Software Engineer, Technical Writer, and Technical Support Engineer actively seeking new opportunities, particularly in areas related to web security, finance, healthcare, and education. If you think my expertise aligns with your team's needs, let's chat! You can find me on LinkedIn and X. I am also an email away.

If you found this article valuable, consider sharing it with your network to help spread the knowledge!

Asynchronous Server: Building and Rigorously Testing a WebSocket and HTTP Server

John Owolabi Idogun — Thu, 20 Feb 2025 06:58:20 +0000

Introduction

Part 5 is already out here: https://johnowolabiidogun.dev/blog/building-an-ai-powered-financial-behavior-analyzer-with-nodejs-python-sveltekit-and-tailwindcss-part-5-dashboard-fc55f4/67b6cf47fe10666dd67e149d

In this article, a detour from the ongoing series on building a financial data analyzer, I focus on the critical aspect of rigorously testing the server. Ensuring appropriate and accurate data handling is paramount. While Python doesn't enforce strict type-safety, I'll demonstrate how to use tools like mypy, bandit, and later prospector to maintain basic code quality and standards.

Prerequisite

To follow along, it's recommended you've read the article on building the preliminary AI service. This article builds upon the concepts and code established there.

Source code

Sirneij / finance-analyzer

An AI-powered financial behavior analyzer and advisor written in Python (aiohttp) and TypeScript (ExpressJS & SvelteKit with Svelte 5)

View on GitHub

Implementation

Step 1: Improving the Initial AI Service

The AI service described in the AI service has several areas for improvement:

Testability: The current structure makes automated testing (both integration and unit) difficult.
Model Accuracy: The zero-shot classification model, originally designed for sentiment analysis, isn't optimal for categorizing financial transactions. A more suitable model is needed.
Code Quality: The code requires refactoring, cleanup, and the addition of new features.
Type Consistency: Type annotations need to be consistently applied and enforced throughout the codebase.

To address these, we will adopt this structure:

.
├── README.md
├── mypy.ini
├── requirements.dev.txt
├── requirements.txt
├── run.py
├── scripts
│   └── test_app.sh
├── src
│   ├── __init_.py
│   ├── app
│   │   ├── __init__.py
│   │   └── app_instance.py
│   ├── models
│   │   ├── __init__.py
│   │   └── base.py
│   └── utils
│       ├── __init__.py
│       ├── analyzer.py
│       ├── base.py
│       ├── extract_text.py
│       ├── resume_parser.py
│       ├── settings.py
│       ├── summarize.py
│       └── websocket.py
└── tests
    ├── __init__.py

We introduced the src/ directory to house the entire application. The aiohttp server setup was refactored into src/app/app_instance.py, with run.py simply responsible for running the created app instance:

import os

from aiohttp import web

from src.app.app_instance import init_app
from src.utils.settings import base_settings

if __name__ == '__main__':
    app = init_app()
    try:
        web.run_app(
            app,
            host='0.0.0.0',
            port=int(os.environ.get('PORT', 5173)),
        )
    except KeyboardInterrupt:
        base_settings.logger.info('Received keyboard interrupt...')
    except Exception as e:
        base_settings.logger.error(f'Server error: {e}')
    finally:
        base_settings.logger.info('Server shutdown complete.')

The run.py file initializes and starts the aiohttp application.

The key changes in app_instance.py are highlighted below:

+ import asyncio
+ from weakref import WeakSet
...
- from utils.analyzer import analyze_transactions
- from utils.extract_text import extract_text_from_pdf
- from utils.resume_parser import extract_text_with_pymupdf, parse_resume_text
- from utils.settings import base_settings
- from utils.summarize import summarize_transactions
- from utils.websocket import WebSocketManager
+ from src.utils.analyzer import analyze_transactions
+ from src.utils.extract_text import extract_text_from_pdf
+ from src.utils.resume_parser import extract_text_with_pymupdf, parse_resume_text
+ from src.utils.settings import base_settings
+ from src.utils.summarize import summarize_transactions
+ from src.utils.websocket import WebSocketManager

# Replace global ws_connections with typed version
- ws_connections: set[WebSocketResponse] = set()
- ws_lock = Lock()
+ WEBSOCKETS = web.AppKey("websockets", WeakSet[WebSocketResponse])


- async def start_background_tasks(app):
+ async def start_background_tasks(app: web.Application) -> None:
    """Initialize application background tasks."""
-   app['ws_connections'] = ws_connections
-   app['ws_lock'] = ws_lock
+   app[WEBSOCKETS] = WeakSet()


- async def cleanup_background_tasks(app):
-     """Cleanup application resources."""
-     await cleanup_ws(app)


- async def cleanup_ws(app):
+ async def cleanup_ws(app: web.Application) -> None:
    """Cleanup WebSocket connections on shutdown."""
-     async with ws_lock:
-         connections = set(ws_connections)  # Create a copy to iterate safely
-         for ws in connections:
-             await ws.close(code=WSMsgType.CLOSE, message='Server shutdown')
-         ws_connections.clear()
+     for websocket in set(app[WEBSOCKETS]):  # type: ignore
+        await websocket.close(code=WSCloseCode.GOING_AWAY, message=b'Server shutdown')

async def websocket_handler(request: Request) -> WebSocketResponse:
    """WebSocket handler for real-time communication."""
    ws = web.WebSocketResponse()
    await ws.prepare(request)

-    async with ws_lock:
-        ws_connections.add(ws)
+   request.app[WEBSOCKETS].add(ws)
    ws_manager = WebSocketManager(ws)
    await ws_manager.prepare()

+    async def ping_server(ws: WebSocketResponse) -> None:
+        try:
+            while True:
+                await ws.ping()
+                await asyncio.sleep(25)
+        except ConnectionResetError:
+            base_settings.logger.info("Client disconnected")
+        finally:
+            await ws.close()
+
+    asyncio.create_task(ping_server(ws))

    base_settings.logger.info('WebSocket connection established')

    try:
        async for msg in ws:
+            if msg.type == WSMsgType.PING:
+                base_settings.logger.info('Intercepted PING from client')
+                await ws.pong(msg.data)
+            elif msg.type == WSMsgType.PONG:
+                base_settings.logger.info('Intercepted PONG from client')
            if msg.type == WSMsgType.TEXT:
                ...
-            elif msg.type == WSMsgType.ERROR:
+            elif msg.type in (WSMsgType.CLOSE, WSMsgType.ERROR):
-                base_settings.logger.error(f'WebSocket error: {ws.exception()}')
+               base_settings.logger.info(
+                    'WebSocket is closing or encountered an error',
+                )
+                break
    except Exception as e:
        base_settings.logger.error(f'WebSocket handler error: {str(e)}')
    finally:
-        async with ws_lock:
-            ws_connections.remove(ws)
-        if not ws.closed:
-            await ws.close()
+        request.app[WEBSOCKETS].discard(ws)
+        if not ws.closed:
+            await ws.close()
        base_settings.logger.info('WebSocket connection closed')

    return ws

def init_app() -> web.Application:
    ...
    # Add startup/cleanup handlers
    app.on_startup.append(start_background_tasks)
-    app.on_cleanup.append(cleanup_background_tasks)
+    app.on_shutdown.append(cleanup_ws)

    return app

We improved type consistency throughout the codebase, using # type: ignore where necessary. We also replaced the global WebSocket connection list with weakref.WeakSet for more robust connection management during shutdown. To maintain persistent connections during long-running processes like zero-shot classification, we implemented a ping/pong mechanism.

Next, we consolidated common utility functions into a new src/utils/base.py file. This included functions like validate_and_convert_transactions, get_device, detect_anomalies, analyze_spending, predict_trends, calculate_trend, and calculate_percentage_change, previously located in utils/summarize.py and utils/analyze.py. We also introduced new functions to estimate financial health (calculate_financial_health) and detect recurring transactions (analyze_recurring_transactions). The anomaly detection was enhanced to identify single-instance anomalies, and the transaction grouping algorithm now uses difflib for fuzzy matching of descriptions. For example, difflib might consider these descriptions to be similar (approximately 69% match): "Target T-12345 Anytown USA" and "Target 12345 Anytown USA":

def group_transactions_by_description(transactions: list[Transaction], cutoff: float = 0.69) -> dict[str, list[float]]:
    """
    Group transactions by description using fuzzy matching with difflib.

    Returns a dictionary mapping a representative description (the group key)
    to a list of transaction amounts. Two descriptions are grouped together if
    their similarity is above a certain threshold.
    """
    groups: dict[str, list[float]] = {}

    for tx in transactions:
        desc = tx.description.lower().strip()
        # Try to find an existing key similar to desc.
        # difflib.get_close_matches returns a list of close matches.
        close_matches = difflib.get_close_matches(desc, groups.keys(), n=1, cutoff=cutoff)
        if close_matches:
            matched_key = close_matches[0]
        else:
            matched_key = None

        if matched_key:
            groups[matched_key].append(tx.amount)
        else:
            groups[desc] = [tx.amount]

    return groups


def find_group_key(description: str, group_keys: list[str], cutoff: float = 0.69) -> str:
    """
    Find the best matching key from group_keys for the given description using difflib.
    Returns the matched key if similarity is above cutoff; otherwise, returns the description.
    """
    desc = description.lower().strip()
    matches = difflib.get_close_matches(desc, group_keys, n=1, cutoff=cutoff)
    if matches:
        return matches[0]
    return desc

We also encapsulated sending progress reports in a reusable function, update_progress.

In src/utils/analyzer.py, the major improvements are:

Improved Model Accuracy: We switched from the yiyanghkust/finbert-tone model to facebook/bart-large-mnli for zero-shot classification. This significantly improves accuracy, although at the cost of speed. For multilingual support, joeddav/xlm-roberta-large-xnli is another option.
Hybrid Classification Approach: We now use a hybrid approach, first attempting to classify transactions using pattern matching. Any remaining unclassified transactions are then processed by the ML model. To improve performance, we process transactions in batches, releasing the event loop after each batch to allow other operations to proceed and to clear memory.
Offloading Calculations: To reduce the load on the classification process, we moved the calculation of anomalies, spending_analysis, spending_trends, recurring_transactions, and financial_health to src/utils/summarize.py, which is significantly faster.

Step 2: Enforcing Type Safety, Security, and Style

Our type annotations are currently only decorative. To enforce type safety, ensure code security, and maintain consistent code style, we'll use the following tools:

mypy: A static type checker.
bandit: A security linter.
black: An uncompromising code formatter.
isort: A tool for sorting imports.

Tip: Consider using Prospector

Prospector provides comprehensive static analysis and ensures your code conforms to PEP8 and other style guidelines. It's highly recommended for in-depth code quality checks.

Install these tools and add them to requirements.dev.txt:

(virtualenv) pip install mypy bandit black isort

Create a mypy.ini file at the root of the project with the following configuration:

# some config from:
# https://www.ralphminderhoud.com/blog/django-mypy-check-runs/
[mypy]
# The mypy configurations: https://mypy.readthedocs.io/en/latest/config_file.html
python_version = 3.13

check_untyped_defs = True
disallow_untyped_defs= True
disallow_incomplete_defs = True
disallow_any_generics = True
disallow_untyped_calls = True
# needs this because celery doesn't have typings
disallow_untyped_decorators = False
ignore_errors = False
ignore_missing_imports = True
implicit_reexport = False
strict_optional = True
strict_equality = True
no_implicit_optional = True
warn_unused_ignores = True
warn_redundant_casts = True
warn_unused_configs = True
warn_unreachable = True
warn_no_return = True
# added these 2 option in mypy 0.800 to enable it to run in our code base
explicit_package_bases = True
namespace_packages = True

[mypy-*.migrations.*]
ignore_errors = True

This configuration enforces various type-checking rules. Each option is generally self-explanatory.

Next, create a bash script (scripts/static_check.sh) to automate the static analysis process:

#!/usr/bin/env bash

set -e

# run black - make sure everyone uses same python style
black --skip-string-normalization --line-length 120 --check src/
black --skip-string-normalization --line-length 120 --check run.py
black --skip-string-normalization --line-length 120 --check tests/

# run isort for import structure checkup with black profile
isort --atomic --profile black -c src/
isort --atomic --profile black -c run.py
isort --atomic --profile black -c tests/

# run mypy
mypy src/

# run bandit - A security linter from OpenStack Security
bandit -r src/

# python static analysis
# prospector  --profile=.prospector.yml --path=src --ignore-patterns=static
# prospector  --profile=.prospector.yml --path=tests --ignore-patterns=static

This script checks the code against the defined standards. To ensure your code passes these checks, run the following commands before committing:

black --skip-string-normalization --line-length 120  src tests *.py

isort --atomic --profile black src tests *.py

To enforce these rules in a team environment, we'll use a CI/CD pipeline. This pipeline runs these checks, and any failure prevents the pull or merge request from being merged. We will use GitHub Actions for our CI/CD. Create a .github/workflows/aiohttp.yml file:

name: UTILITY-SERVER CI

on:
  push:
    branches: [utility]
  pull_request:
    branches: [utility]

jobs:
  build:
    runs-on: ubuntu-latest
    strategy:
      max-parallel: 4
      matrix:
        python-version: [3.13] #[3.7, 3.8, 3.9]

    steps:
      - uses: actions/checkout@v4
      - name: Install system dependencies
        run: |
          sudo apt-get update
          sudo apt-get install -y \
            poppler-utils \
            tesseract-ocr \
            libtesseract-dev \
            libglib2.0-0

      - name: Set up Python ${{ matrix.python-version }}
        uses: actions/setup-python@v3
        with:
          python-version: ${{ matrix.python-version }}
      - name: Install Dependencies
        run: |
          python -m pip install --upgrade pip
          pip install -r requirements.dev.txt
      - name: Debug Environment
        run: |
          python -c "import sys; print(sys.version)"
          python -c "import platform; print(platform.platform())"
          env | sort
      - name: Run static analysis
        run: ./scripts/static_check.sh
      - name: Run Tests
        env:
          LC_ALL: en_NG.UTF-8
          LANG: en_NG.UTF-8
          LABELS: groceries,school,housing,transportation,gadgets,entertainment,utilities,credit cards,other,dining out,healthcare,insurance,savings,investments,childcare,travel,personal care,debts,charity,taxes,subscriptions,streaming services,home maintenance,shopping,pets,fitness,hobbies,gifts
        run: |
          coverage run --parallel-mode -m unittest discover tests && coverage combine && coverage report -m && coverage html

GitHub Actions uses .yaml or .yml files to define workflows, similar to docker-compose.yml. In this case, we're using the latest Ubuntu distribution as the environment. We use version 4 of the actions/checkout action to check out our repository. We also install system dependencies required by some of the Python packages, such as poppler-utils for pdf2image and tesseract-ocr and libtesseract-dev for pytesseract. Since our project doesn't have database interaction, we don't need a services section. The remaining steps are self-explanatory. We then execute our bash script to check the codebase against our defined standards. We also supply environment variables and run the tests (which we'll write later). This CI/CD pipeline runs on every pull request or push to the utility branch.

Step 3: Writing the tests

The last part of our CI/CD was running tests and getting coverage reports. In the Python ecosystem, pytest is an extremely popular testing framework. Though very tempting and might still be used later on, we will stick with Python's built-in testing library, unittest, and use coverage for measuring code test coverage of our program. Let's start with the test setup:

import unittest
import uuid
from datetime import datetime

from aiohttp.test_utils import AioHTTPTestCase

from src.app.app_instance import init_app
from src.utils.websocket import WebSocketManager


class Base:
    """Base class for tests."""

    def create_transaction_dict(
        self,
        date: datetime | str,
        description: str,
        amount: float,
        balance: float,
        type_str='expense',
        include_v: bool = True,
    ) -> dict:
        txn = {
            '_id': str(uuid.uuid4()),
            'date': date.isoformat() if isinstance(date, datetime) else date,
            'createdAt': date.isoformat() if isinstance(date, datetime) else date,
            'updatedAt': date.isoformat() if isinstance(date, datetime) else date,
            'description': description,
            'amount': amount,
            'balance': balance,
            'type': type_str,
            'userId': '1',
        }
        if include_v:
            txn['__v'] = 0

        return txn


# A simple fake WebSocketResponse to simulate aiohttp behavior.
class FakeWebSocket:
    def __init__(self, raise_on_send=False):
        self.messages = []  # will store the JSON messages sent
        self.closed = False
        self.raise_on_send = raise_on_send
        self.close_code = None
        self.close_message = None

    async def send_json(self, data):
        if self.raise_on_send:
            raise Exception('send_json error')
        self.messages.append(data)

    async def close(self, code=None, message=None):
        self.closed = True
        self.close_code = code
        self.close_message = message


class BaseAsyncTestClass(Base, unittest.IsolatedAsyncioTestCase):
    """Base class for async tests."""

    async def asyncSetUp(self):
        # Create a FakeWebSocket for each test.
        self.fake_ws = FakeWebSocket()
        self.websocket_manager = WebSocketManager(self.fake_ws)


class BaseTestClass(Base, unittest.TestCase):
    """Base class for sync tests."""


class BaseAioHTTPTestCase(Base, AioHTTPTestCase):
    """Base class for aiohttp tests."""

    async def get_application(self):
        return init_app()

    async def asyncSetUp(self):
        await super().asyncSetUp()
        # Create a FakeWebSocket for each test.
        self.fake_ws = FakeWebSocket()
        self.websocket_manager = WebSocketManager(self.fake_ws)


if __name__ == '__main__':
    unittest.main()

We simply have classes which provide blueprints for our tests. The Base class makes the create_transaction_dict method available to all its children, simplifying the creation of transaction data for tests. The FakeWebSocket class simulates aiohttp WebSocket behavior, which is essential for unit testing the project's WebSocket utilities. All asynchronous unit tests inherit from BaseAsyncTestClass, while synchronous tests inherit from BaseTestClass. BaseAioHTTPTestCase is used for integration-style tests that involve the aiohttp application. The get_application is required in this class to return our app's instance.

Note: Unit vs Integration tests

A unit test focuses on testing a single piece of code (like a function such as analyze_recurring_transactions) whereas integration tests examine how multiple units of code interact with each other within a system (this is like testing the behavior of sending a request to /ws)

Let's take an example integration-style test, especially for our websocket, and another unit test for some of the subprocesses to balance things out:

import asyncio
import json
from unittest.mock import AsyncMock, patch

from aiohttp import WSMsgType

from src.app.app_instance import WEBSOCKETS
from tests import BaseAioHTTPTestCase


class TestWebSocketHandler(BaseAioHTTPTestCase):
    """Exhaustively test the WebSocket handler."""

    async def setUpAsync(self):
        await super().setUpAsync()
        # Capture the original create_task function.
        self.orig_create_task = asyncio.create_task

    async def __dummy_analyze(self, transactions, ws_manager):
        """Dummy analyze implementation that returns a known result."""
        return {
            'categories': {
                'expenses': {
                    'groceries': 10.0,
                    'rent': 90.0,
                },
                'expense_percentages': {
                    'groceries': 5,
                    'rent': 45,
                },
                'income': 200.0,
            }
        }

    async def __dummy_summarize(self, transactions, ws_manager):
        """Dummy summarize implementation that returns a known result."""
        return {
            'income': {
                'total': 200.00,
                'trend': 'neutral',
                'change': 0.0,
            },
            'expenses': {
                'total': 100.00,
                'trend': 'neutral',
                'change': 0.0,
            },
            'savings': {
                'total': 100.00,
                'trend': 'neutral',
                'change': 0.0,
            },
            'total_transactions': 2,
            'expense_count': 1,
            'income_count': 1,
            'avg_expense': 100.00,
            'avg_income': 200.00,
            'start_date': '2022-01-01',
            'end_date': '2022-01-31',
            'largest_expense': 200.00,
            'largest_income': 200.00,
            'savings_rate': 50.0,
            'monthly_summary': {
                '2022-01': {
                    'income': 200.00,
                    'expenses': 100.00,
                    'savings': 100.00,
                },
            },
            'anomalies': [],
            'spending_analysis': {
                'total_spent': 100.00,
                'total_income': 200.00,
                'savings_rate': 50.0,
                'daily_summary': {
                    '2022-01-01': {
                        'total_spent': 100.00,
                        'total_income': 200.00,
                        'savings_rate': 50.0,
                    },
                },
                'cumulative_balance': {
                    '2022-01-01': 100.00,
                },
            },
            'spending_trends': {
                'total_spent': 100.00,
                'total_income': 200.00,
                'savings_rate': 50.0,
            },
            'recurring_transactions': [],
            'financial_health': {
                'debt_to_income_ratio': 0,
                'savings_rate': 0,
                'balance_growth_rate': 0,
                'financial_health_score': 0,
            },
        }

    def __dummy_create_task(self, coro):
        if hasattr(coro, 'cr_code') and 'ping_server' in coro.cr_code.co_qualname:
            # Explicitly close the ping_server coroutine so it doesn't leak.
            coro.close()
            # Return a dummy, already‐completed future.
            fut = asyncio.Future()
            fut.set_result(None)
            return fut
        return self.orig_create_task(coro)

    async def __receive_messages(self, ws, count, timeout=5):
        """Helper to collect 'count' text messages from the WebSocket."""
        messages = []
        while len(messages) < count:
            msg = await ws.receive(timeout=timeout)
            if msg.type == WSMsgType.TEXT:
                messages.append(json.loads(msg.data))
            elif msg.type == WSMsgType.CLOSE:
                break
        return messages

    async def test_analyze_action(self):
        """Test that sending an 'analyze' action yields progress and result messages."""
        self.transactions = [
            self.create_transaction_dict('2022-01-01', 'Transaction 1', -100.0, 100.0),
            self.create_transaction_dict('2022-01-02', 'Transaction 2', 200.0, 300.0),
        ]

        # Patch the analyzer so that it returns a predictable result,
        # and patch create_task with our dummy version.
        with patch(
            'src.app.app_instance.analyze_transactions',
            new=AsyncMock(side_effect=self.__dummy_analyze),
        ), patch("asyncio.create_task", self.__dummy_create_task):
            ws = await self.client.ws_connect('/ws')
            msg_data = {'action': 'analyze', 'transactions': self.transactions}
            await ws.send_str(json.dumps(msg_data))
            # This helps avoid timout errors when the server is slow to respond.
            messages = await self.__receive_messages(ws, 2)
            self.assertEqual(len(messages), 2)

            # First response: progress message.
            progress, result = messages
            self.assertEqual(progress.get('action'), 'progress')
            self.assertEqual(progress.get('message'), 'Analysis complete')
            self.assertEqual(progress.get('progress'), 1.0)
            self.assertEqual(progress.get('taskType'), 'Analysis')

            # Second response: result message.
            self.assertEqual(result.get('action'), 'analysis_complete')
            self.assertEqual(result.get('taskType'), 'Analysis')
            expected_data = await self.__dummy_analyze(self.transactions, self.websocket_manager)
            self.assertEqual(result.get('result'), expected_data)
            await ws.close()

    async def test_summary_action(self):
        """Test that sending a 'summary' action yields progress and result messages."""
        self.transactions = [
            self.create_transaction_dict('2022-01-01', 'Transaction 1', -100.0, 100.0),
            self.create_transaction_dict('2022-01-02', 'Transaction 2', 200.0, 300.0),
        ]

        # Patch the summarizer so that it returns a predictable result,
        # and patch create_task with our dummy version.
        with patch(
            'src.app.app_instance.summarize_transactions',
            new=AsyncMock(side_effect=self.__dummy_summarize),
        ), patch("asyncio.create_task", self.__dummy_create_task):
            ws = await self.client.ws_connect('/ws')
            msg_data = {'action': 'summary', 'transactions': self.transactions}
            await ws.send_str(json.dumps(msg_data))
            # This helps avoid timout errors when the server is slow to respond.
            messages = await self.__receive_messages(ws, 2)
            self.assertEqual(len(messages), 2)

            # First response: progress message.
            progress, result = messages
            self.assertEqual(progress.get('action'), 'progress')
            self.assertEqual(progress.get('message'), 'Summary complete')
            self.assertEqual(progress.get('progress'), 1.0)
            self.assertEqual(progress.get('taskType'), 'Summarize')

            # Second response: result message.
            self.assertEqual(result.get('action'), 'summary_complete')
            self.assertEqual(result.get('taskType'), 'Summarize')
            expected_data = await self.__dummy_summarize(self.transactions, self.websocket_manager)
            self.assertEqual(result.get('result'), expected_data)
            await ws.close()

    async def test_unknown_action(self):
        """Test that an unknown action returns an error message."""
        ws = await self.client.ws_connect('/ws')
        msg_data = {'action': 'nonexistent'}
        await ws.send_str(json.dumps(msg_data))
        # This helps avoid timout errors when the server is slow to respond.
        messages = await self.__receive_messages(ws, 1)
        self.assertEqual(len(messages), 1)
        error = messages[0]
        self.assertEqual(error.get('action'), 'error')
        self.assertEqual(error.get('taskType'), 'Error')
        self.assertEqual(error.get('result'), {'message': 'Unknown action'})
        await ws.close()

    async def test_message_processing_exception(self):
        """Test that sending invalid JSON produces an error message."""
        ws = await self.client.ws_connect('/ws')
        await ws.send_str('invalid json')
        # This helps avoid timout errors when the server is slow to respond.
        messages = await self.__receive_messages(ws, 1)
        self.assertEqual(len(messages), 1)
        error = messages[0]
        self.assertEqual(error.get('action'), 'error')
        self.assertEqual(error.get('taskType'), 'Error')
        self.assertEqual(error.get('result'), {'error': 'Expecting value: line 1 column 1 (char 0)'})
        await ws.close()

    async def test_close_on_error(self):
        """Test that when a client closes the connection, the WebSocket is removed from the app."""
        ws = await self.client.ws_connect('/ws')
        await ws.send_str('invalid json')
        # This helps avoid timout errors when the server is slow to respond.
        messages = await self.__receive_messages(ws, 1)
        self.assertEqual(len(messages), 1)
        await ws.close()
        self.assertNotIn(ws, self.app[WEBSOCKETS])

Overlooking the dummy data generators, the __receive_messages helper is crucial for accumulating WebSocket messages. Without it, attempting await ws.receive_json(...) multiple times could lead to timeout errors, resulting in cryptic tracebacks:

----------------------------------------------------------------------
Traceback (most recent call last):
  File ".../utility/virtualenv/lib/python3.13/site-packages/aiohttp/client_ws.py", line 332, in receive
    msg = await self._reader.read()
          ^^^^^^^^^^^^^^^^^^^^^^^^^
  File "aiohttp/_websocket/reader_c.py", line 109, in read
  File "aiohttp/_websocket/reader_c.py", line 106, in aiohttp._websocket.reader_c.WebSocketDataQueue.read
asyncio.exceptions.CancelledError

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "/opt/homebrew/Cellar/python@3.13/3.13.2/Frameworks/Python.framework/Versions/3.13/lib/python3.13/asyncio/runners.py", line 118, in run
    return self._loop.run_until_complete(task)
           ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~^^^^^^
  File "/opt/homebrew/Cellar/python@3.13/3.13.2/Frameworks/Python.framework/Versions/3.13/lib/python3.13/asyncio/base_events.py", line 725, in run_until_complete
    return future.result()
           ~~~~~~~~~~~~~^^
  File ".../utility/tests/app/websocket_handler/test_integration.py", line 114, in __receive_messages
    msg = await ws.receive_json(timeout=timeout)
          ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File ".../utility/virtualenv/lib/python3.13/site-packages/aiohttp/client_ws.py", line 331, in receive
    async with async_timeout.timeout(receive_timeout):
               ~~~~~~~~~~~~~~~~~~~~~^^^^^^^^^^^^^^^^^
  File "/opt/homebrew/Cellar/python@3.13/3.13.2/Frameworks/Python.framework/Versions/3.13/lib/python3.13/asyncio/timeouts.py", line 116, in __aexit__
    raise TimeoutError from exc_val
TimeoutError

The helper also aids in filtering messages of interest. We also created a dummy version of ping_server to properly close it and prevent memory leaks. With the dummy functions in place, we created test cases that interact with our WebSocket endpoint. Using async patches and mocks, we fed predictable responses to the tests. Note that we used our async dummy methods as the side_effect of the AsyncMock. Using return_value instead of side_effect in the mocks prolonged the processes and caused timeout errors.

The other test cases handle various scenarios to provide better test coverage.

Warning: The file path in patch

When supplying file paths in patch, use the path where the program is used, not where it was defined. For instance, src.app.app_instance.analyze_transactions was defined in src/utils/analyzer.py but since it was used in src/app/app_instance.py, we used src.app.app_instance.analyze_transactions.

However, the integration testing approach poses some limitations. We can't modify the internals of the aiohttp WebSocket instance. This is where unit testing comes to the rescue, as we can modify internals and mock them as needed to thoroughly test the desired feature. Hence the other test file for our WebSocket, tests/app/websocket_handler/test_ping.py.

To wrap up, let's see how we tested the src/utils/analyze.py:

import uuid
from datetime import datetime, timedelta
from unittest.mock import patch

import torch

from src.utils.analyzer import analyze_transactions, classify_transactions
from src.utils.base import (
    analyze_recurring_transactions,
    analyze_spending,
    calculate_financial_health,
    detect_anomalies,
    predict_trends,
    validate_and_convert_transactions,
)
from tests import BaseAsyncTestClass


class TestAnalyzer(BaseAsyncTestClass):
    @patch(
        'src.utils.analyzer.pipeline', return_value=lambda *args, **kwargs: [{'labels': ['groceries'], 'scores': [1.0]}]
    )
    async def test_analyze_transactions_valid(self, mock_pipeline):
        tx_data = [
            {
                '_id': str(uuid.uuid4()),
                'date': '2024-01-01T00:00:00',
                'createdAt': '2024-01-01T00:00:00',
                'updatedAt': '2024-01-01T00:00:00',
                'description': 'Test expense',
                'amount': -100,
                'balance': 900,
                'type': 'expense',
                'userId': '1',
            },
            {
                '_id': str(uuid.uuid4()),
                'date': '2024-01-02T00:00:00',
                'createdAt': '2024-01-02T00:00:00',
                'updatedAt': '2024-01-02T00:00:00',
                'description': 'Salary',
                'amount': 2000,
                'balance': 2900,
                'type': 'income',
                'userId': '1',
            },
        ]
        result = await analyze_transactions(tx_data)
        self.assertIn('categories', result)
    ...

    async def test_classify_transactions_pattern_matching(self):
        """
        Test that transactions with descriptions matching common patterns
        are categorized correctly without invoking the ML pipeline.
        """
        # Create dummy transactions that should match predefined patterns
        tx1 = self.create_transaction_dict('2024-01-01T00:00:00', 'Walmart grocery purchase', -50.0, 950.0, 'expense')
        tx2 = self.create_transaction_dict('2024-01-02T00:00:00', 'Uber ride', -20.0, 930.0, 'expense')
        tx3 = self.create_transaction_dict('2024-01-03T00:00:00', 'Netflix subscription', -15.0, 915.0, 'expense')
        tx4 = self.create_transaction_dict('2024-01-04T00:00:00', 'Salary', 3000.0, 3915.0, 'income')

        # We assume pattern matching is applied first
        transactions = await validate_and_convert_transactions([tx1, tx2, tx3, tx4])
        # Call classify_transactions without a WebSocket manager
        result = await classify_transactions(transactions)
        categories = result.get('expenses', {})
        income_total = result.get('income', 0)

        # Check that the descriptions are mapped to expected categories:
        # "Walmart grocery" should fall under 'groceries'
        self.assertIn('groceries', categories)
        self.assertGreater(categories['groceries'], 0)

        # "Uber ride" should fall under 'transportation'
        self.assertIn('transportation', categories)
        self.assertGreater(categories['transportation'], 0)

        # "Netflix subscription" should be captured under 'subscriptions'
        self.assertIn('subscriptions', categories)
        self.assertGreater(categories['subscriptions'], 0)

        # Income should include the salary
        self.assertEqual(income_total, 3000)

    @patch.dict(
        'os.environ',
        {"LABELS": "groceries,housing,transportation,entertainment,utilities,education,credit_cards,insurance,other"},
    )
    @patch('src.utils.analyzer.pipeline')
    async def test_classify_transactions_ml_fallback(self, mock_pipeline):
        # Simulate a transaction with an unmatched description
        tx1 = self.create_transaction_dict(
            '2024-01-05T00:00:00', 'Unusual expense with no pattern', -75.0, 840.0, 'expense'
        )

        # Setup the fake pipeline result
        fake_result = [{'labels': ['other'], 'scores': [0.95]}]
        mock_pipeline.return_value = lambda *args, **kwargs: fake_result

        transactions = await validate_and_convert_transactions([tx1])
        result = await classify_transactions(transactions)
        categories = result.get('expenses', {})

        # Expect that the ML fallback has assigned this expense to 'other'
        self.assertIn('other', categories)
        self.assertAlmostEqual(categories['other'], 75 * 0.95, places=2)

    async def test_analyze_recurring_transactions_monthly(self):
        """
        Test that transactions with the same description and a roughly monthly interval
        are detected as recurring.
        """

        base_date = datetime(2024, 1, 1)
        # Create 3 monthly transactions (interval ~30 days)
        tx1 = self.create_transaction_dict((base_date).isoformat(), 'Gym membership', -50.0, 950.0, 'expense')
        tx2 = self.create_transaction_dict(
            (base_date + timedelta(days=30)).isoformat(), 'Gym membership', -50.0, 900.0, 'expense'
        )
        tx3 = self.create_transaction_dict(
            (base_date + timedelta(days=60)).isoformat(), 'Gym membership', -50.0, 850.0, 'expense'
        )

        transactions = await validate_and_convert_transactions([tx1, tx2, tx3])

        recurring = analyze_recurring_transactions(transactions)
        self.assertTrue(len(recurring) > 0)
        monthly_recurring = next((r for r in recurring if r['frequency'] == 'Monthly'), None)
        self.assertIsNotNone(monthly_recurring)
        self.assertEqual(monthly_recurring['description'], 'gym membership')

    async def test_analyze_recurring_transactions_weekly(self):
        """
        Test that transactions with the same description and a roughly weekly interval
        are detected as recurring.
        """

        base_date = datetime(2024, 1, 1)
        # Create 3 weekly transactions (interval ~7 days)
        tx1 = self.create_transaction_dict((base_date).isoformat(), 'Weekly yoga class', -20.0, 980.0, 'expense')
        tx2 = self.create_transaction_dict(
            (base_date + timedelta(days=7)).isoformat(), 'Weekly yoga class', -20.0, 960.0, 'expense'
        )
        tx3 = self.create_transaction_dict(
            (base_date + timedelta(days=14)).isoformat(), 'Weekly yoga class', -20.0, 940.0, 'expense'
        )

        transactions = await validate_and_convert_transactions([tx1, tx2, tx3])

        recurring = analyze_recurring_transactions(transactions)
        self.assertTrue(len(recurring) > 0)
        weekly_recurring = next((r for r in recurring if r['frequency'] == 'Weekly'), None)
        self.assertIsNotNone(weekly_recurring)
        self.assertEqual(weekly_recurring['description'], 'weekly yoga class')

    ...

    def test_edge_empty_transactions(self):
        """
        Ensure that functions gracefully handle an empty list of transactions.
        """
        # predict_trends should return a message indicating insufficient data
        trends = predict_trends([])
        self.assertIn('trend', trends)
        self.assertEqual(trends['trend'], 'Not enough data')

        # calculate_financial_health on empty list should not crash (might return infinity or 0)
        health = calculate_financial_health([])
        self.assertIn('debt_to_income_ratio', health)
        self.assertIn('savings_rate', health)
        self.assertIn('balance_growth_rate', health)
        self.assertIn('financial_health_score', health)

    async def test_not_transaction_analyzer(self):
        """
        Ensure that functions gracefully handle invalid transaction data.
        """
        analysis = await analyze_transactions(None, self.websocket_manager)
        self.assertIn('error', analysis)
        self.assertEqual(analysis['error'], 'No transactions provided')
        self.assertTrue(self.fake_ws.messages)

    async def test_analyze_transactions_with_websocket(self):
        """Test the analyze_transactions function with a WebSocketManager."""
        # Create an invalid transaction
        tx_1 = [
            {
                '_id': str(uuid.uuid4()),
                'date': '2024-01-01T00:00:00',
                'createdAt': '2024-01-01T00:00:00',
                'updatedAt': '2024-01-01T00:00:00',
                'description': 'Test expense',
                'amount': -100,
                'type': 'expense',
                'userId': '1',
            }
        ]

        result = await analyze_transactions(tx_1, self.websocket_manager)
        self.assertIn('error', result)
        self.assertEqual(result['error'], 'No valid transactions provided')
        # Check that progress messages were sent
        self.assertTrue(self.fake_ws.messages)

        # Create valid transactions
        tx_2 = self.create_transaction_dict('2024-01-01T00:00:00', 'Salary', 2000, 2900, 'income')
        result = await analyze_transactions([tx_2], self.websocket_manager)
        self.assertIn('categories', result)
        # Check that progress messages were sent
        self.assertTrue(self.fake_ws.messages)

    @patch('src.utils.analyzer.validate_and_convert_transactions')
    async def test_analyze_transactions_validation_exception(self, mock_validate):
        """Test analyze_transactions handling when validation fails"""
        valid_tx = self.create_transaction_dict('2024-01-01T00:00:00', 'Salary', 2000, 2900, 'income')
        mock_validate.side_effect = ValueError('Mock validation error')
        result = await analyze_transactions(valid_tx, self.websocket_manager)
        self.assertIn('error', result)
        msg = self.fake_ws.messages[-1]
        self.assertEqual(msg['action'], 'progress')
        self.assertEqual(msg['message'], 'Analysis failed')
        self.assertEqual(msg['taskType'], 'Analysis')

    @patch('src.utils.analyzer.pipeline')
    async def test_classify_transactions_exception(self, mock_pipeline):
        """Test classify_transactions handling when pipeline fails"""
        tx = self.create_transaction_dict('2024-01-01T00:00:00', 'Test expense', -100, 900, 'expense')
        transactions = await validate_and_convert_transactions([tx])
        mock_pipeline.side_effect = RuntimeError('Mock pipeline error')

        result = await classify_transactions(transactions, self.websocket_manager)

        # Check error response
        self.assertIn('error', result)
        self.assertTrue('Classification failed' in result['error'])

        # Check websocket message
        msg = self.fake_ws.messages[-1]
        self.assertEqual(msg['action'], 'progress')
        self.assertEqual(msg['message'], 'Analysis failed')
        self.assertEqual(msg['taskType'], 'Analysis')

    @patch('src.utils.analyzer.get_device')
    @patch('src.utils.analyzer.pipeline')
    async def test_classify_transactions_device_cpu(self, mock_pipeline, mock_device):
        """Test that classify_transactions uses CPU device for the pipeline."""
        tx = self.create_transaction_dict('2024-01-01T00:00:00', 'Test expense', -100, 900, 'expense')
        transactions = await validate_and_convert_transactions([tx])
        mock_device.return_value = (torch.device('cpu'), 'CPU')
        await classify_transactions(transactions, self.websocket_manager)
        mock_pipeline.assert_called_once_with('zero-shot-classification', model='facebook/bart-large-mnli', device=-1)

    @patch('src.utils.analyzer.get_device')
    @patch('src.utils.analyzer.pipeline')
    async def test_classify_transactions_device_gpu(self, mock_pipeline, mock_device):
        """Test that classify_transactions uses GPU device for the pipeline."""
        tx = self.create_transaction_dict('2024-01-01T00:00:00', 'Test expense', -100, 900, 'expense')
        transactions = await validate_and_convert_transactions([tx])
        mock_device.return_value = (torch.device('cuda'), 'GPU')
        await classify_transactions(transactions, self.websocket_manager)
        mock_pipeline.assert_called_once_with('zero-shot-classification', model='facebook/bart-large-mnli', device=0)

    @patch('src.utils.analyzer.get_device')
    @patch('src.utils.analyzer.pipeline')
    async def test_classify_transactions_device_mps(self, mock_pipeline, mock_device):
        """Test that classify_transactions uses MPS (Apple Metal) device for the pipeline."""
        tx = self.create_transaction_dict('2024-01-01T00:00:00', 'Test expense', -100, 900, 'expense')
        transactions = await validate_and_convert_transactions([tx])
        mock_device.return_value = (torch.device('mps'), 'MPS (Apple Metal)')
        await classify_transactions(transactions, self.websocket_manager)
        mock_pipeline.assert_called_once_with('zero-shot-classification', model='facebook/bart-large-mnli', device=0)

This thorough testing allows us to have confidence in the reliability of our code. The repository's tests folder contains other test files that rigorously test our implementations. Currently, we have 100% test coverage on the AI service, and static analysis is enforced.

We will stop here. In the next article, we will return to implementing the dashboard.

Outro

Enjoyed this article? I'm a Software Engineer, Technical Writer and Technical Support Engineer actively seeking new opportunities, particularly in areas related to web security, finance, healthcare, and education. If you think my expertise aligns with your team's needs, let's chat! You can find me on LinkedIn and X. I am also an email away.

If you found this article valuable, consider sharing it with your network to help spread the knowledge!

Building an AI-powered Financial Behavior Analyzer with NodeJS, Python, SvelteKit, and TailwindCSS - Part 4

John Owolabi Idogun — Sat, 15 Feb 2025 15:18:41 +0000

Introduction

Part 5 is already out on: https://johnowolabiidogun.dev/blog/asynchronous-server-building-and-rigorously-testing-a-websocket-and-http-server-3918df/67b0ab3c7a900ac23e502c51

In this part of the series, we will implement our backend service's WebSocket handler and begin building the frontend. This will enable real-time communication between the backend and frontend, allowing us to display analysis results as they become available.

Prerequisite

The main prerequisite is that you have gone through the previous articles in this series. This ensures you have the necessary context and environment set up.

Source code

Sirneij / finance-analyzer

An AI-powered financial behavior analyzer and advisor written in Python (aiohttp) and TypeScript (ExpressJS & SvelteKit with Svelte 5)

View on GitHub

Implementation

Step: WebSocket handler

To provide real-time updates for financial data analysis and summaries, we use WebSockets, a bidirectional communication protocol, instead of traditional HTTP requests that require constant refreshing. The TransactionWebSocketHandler function manages WebSocket connections:

import { WebSocket } from "ws";
import { TransactionService } from "$services/transaction.service.js";
import { baseConfig } from "$config/base.config.js";
import mongoose from "mongoose";
import { sendError } from "$utils/error.utils.js";

export function TransactionWebSocketHandler(ws: WebSocket): void {
  ws.on("message", async (message: string) => {
    try {
      const actions = JSON.parse(message);

      if (!Array.isArray(actions)) {
        sendError(ws, "Invalid message format. Expected an array.");
        return;
      }

      for (const actionObj of actions) {
        if (!actionObj.action || !actionObj.userId) {
          sendError(
            ws,
            "Invalid action format. Each action requires 'action' and 'userId'."
          );
          return;
        }

        const { action, userId } = actionObj;

        if (!mongoose.Types.ObjectId.isValid(userId)) {
          sendError(ws, "Invalid userId format.");
          return;
        }

        switch (action) {
          case "analyze":
          case "summary":
            await handleAction(ws, new mongoose.Types.ObjectId(userId), action);
            break;
          default:
            sendError(ws, `Unknown action: ${action}`);
        }
      }
    } catch (error) {
      baseConfig.logger.error(
        `Error processing message: ${
          error instanceof Error ? error.message : error
        }`
      );
      sendError(
        ws,
        `Failed to process message: ${
          error instanceof Error ? error.message : error
        }`
      );
    }
  });

  ws.on("close", () => {
    baseConfig.logger.info("Frontend WebSocket connection closed");
  });

  ws.on("error", (error) => {
    baseConfig.logger.error(`WebSocket error: ${error.message}`);
  });
}

async function handleAction(
  frontendWs: WebSocket,
  userId: mongoose.Types.ObjectId,
  action: string
) {
  try {
    const transactions = await TransactionService.findTransactionsByUserId(
      userId,
      -1,
      -1
    );

    if (!transactions) {
      sendError(
        frontendWs,
        `No transactions found for userId: ${userId}`,
        action
      );
      return;
    }

    await TransactionService.connectToUtilityServer(
      action,
      transactions.transactions,
      frontendWs
    );
  } catch (error) {
    baseConfig.logger.error(
      `Error handling action: ${error instanceof Error ? error.message : error}`
    );
    sendError(
      frontendWs,
      `Failed to handle action: ${
        error instanceof Error ? error.message : error
      }`
    );
  }
}

Using the ws module (installed in the previous article), this function sets up the WebSocket connection and defines handlers for incoming messages, connection close, and errors. As soon as it receives a message, the callback for the message event listener gets fired and it works as follows:

Parses incoming messages as JSON, expecting an array of actions.
Validates the format of each action, ensuring it contains action and userId.
Validates the userId to ensure it is a valid Mongoose ObjectId.
Uses a switch statement to handle different actions (analyze and summary), and
Calls the handleAction function to process each valid action.

The handleAction processes specific actions requested by the client by retrieving transactions for the specified userId using TransactionService.findTransactionsByUserId static method discussed in the previous article. Any transactions retrieved are sent via the TransactionService.connectToUtilityServer method to the utility server for analysis or summary.

Let's register this handler with our app:

...
import { WebSocketServer } from "ws";
...
import { TransactionWebSocketHandler } from "$websockets/transaction.websocket.js";
...
const startServer = async () => {
  try {
    const server: HttpServer = createServer(app);
    const wss = new WebSocketServer({ server, path: "/ws" });

    // 5. Setup WebSocket handlers
    wss.on("connection", (ws) => {
      TransactionWebSocketHandler(ws);
    });

    // 6. Connect to MongoDB
    baseConfig.logger.info("Connecting to MongoDB cluster...");
    const db = await connectToCluster();

    ...
  } catch (error) {
    baseConfig.logger.error("Error starting server:", error);
    process.exit(1);
  }
};
startServer();

With that, we conclude the backend service. Now it's time to set SvelteKit up with TailwindCSS.

Step 2: SvelteKit with TailwindCSS

To setup a SvelteKit project with TailwindCSS, we will refer to the official TailwindCSS guide with some modifications from the migration guide I previously wrote.

First, create a new SvelteKit project via the Svelte 5 sv CLI:

$ npx sv create interface # modify the name as you please

You will be prompted to install sv which you should accent to. Your interaction with the CLI should look like this:

projects npx sv create interface
Need to install the following packages:
sv@0.6.18
Ok to proceed? (y) y

┌  Welcome to the Svelte CLI! (v0.6.18)
│
◇  Which template would you like?
│  SvelteKit minimal
│
◇  Add type checking with Typescript?
│  Yes, using Typescript syntax
│
◆  Project created
│
◇  What would you like to add to your project? (use arrow keys / space bar)
│  prettier, eslint, vitest, tailwindcss, sveltekit-adapter
│
◇  tailwindcss: Which plugins would you like to add?
│  typography, forms
│
◇  sveltekit-adapter: Which SvelteKit adapter would you like to use?
│  node
│
◇  Which package manager do you want to install dependencies with?
│  npm
│
◆  Successfully setup add-ons
│
◆  Successfully installed dependencies
│
◇  Successfully formatted modified files
│
◇  Project next steps ─────────────────────────────────────────────────────╮
│                                                                          │
│  1: cd interface                                                         │
│  2: git init && git add -A && git commit -m "Initial commit" (optional)  │
│  3: npm run dev -- --open                                                │
│                                                                          │
│  To close the dev server, hit Ctrl-C                                     │
│                                                                          │
│  Stuck? Visit us at https://svelte.dev/chat                              │
│                                                                          │
├──────────────────────────────────────────────────────────────────────────╯
│
└  You're all set!

Feel free to modify any of the steps as you like. You can change directory into the newly created project and install the dependencies.

For some reasons, the tailwindcss installed by sv was version 3.4.17. However, at the time of writting this, TailwindCSS is already at version 4.0.2. So we need to migrate. To incept the migration steps, run this command:

interface$ npx @tailwindcss/upgrade@next

You should get something like this:

interface$ npx @tailwindcss/upgrade@next
Need to install the following packages:
@tailwindcss/upgrade@4.0.2
Ok to proceed? (y) y

≈ tailwindcss v4.0.2

fatal: not a git repository (or any of the parent directories): .git
│ Searching for CSS files in the current
│ directory and its subdirectories…

│ ↳ Linked `./tailwind.config.ts` to
│   `./src/app.css`

│ Migrating JavaScript configuration files…

│ ↳ Migrated configuration file:
│   `./tailwind.config.ts`

│ Migrating templates…

│ ↳ Migrated templates for configuration file:
│   `./tailwind.config.ts`

│ Migrating stylesheets…

│ ↳ Migrated stylesheet: `./src/app.css`

│ Migrating PostCSS configuration…

│ ↳ Installed package: `@tailwindcss/postcss`

│ ↳ Removed package: `autoprefixer`

│ ↳ Migrated PostCSS configuration:
│   `./postcss.config.js`

│ Updating dependencies…

│ ↳ Updated package:
│   `prettier-plugin-tailwindcss`

│ ↳ Updated package: `tailwindcss`

fatal: not a git repository (or any of the parent directories): .git
│ No changes were made to your repository.

It will magically modify your src/app.css to look like:

@import "tailwindcss";

@plugin '@tailwindcss/typography';
@plugin '@tailwindcss/forms';

/*
  The default border color has changed to `currentColor` in Tailwind CSS v4,
  so we've added these compatibility styles to make sure everything still
  looks the same as it did with Tailwind CSS v3.

  If we ever want to remove these styles, we need to add an explicit border
  color utility to any element that depends on these defaults.
*/
@layer base {
  *,
  ::after,
  ::before,
  ::backdrop,
  ::file-selector-button {
    border-color: var(--color-gray-200, currentColor);
  }
}

tailwind.config.ts will be removed and postcss.config.js will be modified. We need to make a slight change to src/app.css and vite.config.js:

@import "tailwindcss";

@plugin '@tailwindcss/typography';
@plugin '@tailwindcss/forms';

@custom-variant dark (&:where(.dark, .dark *));

@layer base {
  *,
  ::after,
  ::before,
  ::backdrop,
  ::file-selector-button {
    border-color: var(--color-gray-200, currentColor);
  }
}

We need line 6 so that we can leverage classes (and later, prefers-color-scheme) to dynamically switch themes.

Next is vite.config.js:

import { defineConfig } from "vitest/config";
import { sveltekit } from "@sveltejs/kit/vite";
import tailwindcss from "@tailwindcss/vite";

export default defineConfig({
  plugins: [tailwindcss(), sveltekit()],

  test: {
    include: ["src/**/*.{test,spec}.{js,ts}"],
  },
});

That concludes the initial setup.

Step 3: Leverage `prefers-color-scheme` in theme toggling

Many modern operating systems (OS) have made dark mode a first-class feature, and tuning your web application to honor the user's OS theme preference provides a better user experience. The CSS prefers-color-scheme media feature makes this easy to implement. Here's how to leverage it:

<!DOCTYPE html>
<html lang="en">
  <head>
    <meta charset="utf-8" />
    <meta name="language" content="en" />
    <!-- Theme -->
    <meta
      name="theme-color"
      content="#ffffff"
      media="(prefers-color-scheme: light)"
    />
    <meta
      name="theme-color"
      content="#111827"
      media="(prefers-color-scheme: dark)"
    />
    <meta name="viewport" content="width=device-width, initial-scale=1" />
    <link rel="preconnect" href="https://fonts.googleapis.com" />
    <link rel="preconnect" href="https://fonts.gstatic.com" crossorigin />
    <link
      href="https://fonts.googleapis.com/css2?family=Fira+Code:wght@300..700&family=Fira+Sans:ital,wght@0,100;0,200;0,300;0,400;0,500;0,600;0,700;0,800;0,900;1,100;1,200;1,300;1,400;1,500;1,600;1,700;1,800;1,900&display=swap"
      rel="stylesheet"
    />
    ... %sveltekit.head%
    <script>
      // Get user's explicit preference
      const userTheme = localStorage.getItem("theme");

      // Get system preference
      const systemTheme = window.matchMedia("(prefers-color-scheme: dark)");

      // Use user preference if set, otherwise follow system
      const theme = userTheme || (systemTheme.matches ? "dark" : "light");

      // Apply theme
      document.documentElement.classList.toggle("dark", theme === "dark");

      // Listen for system changes
      systemTheme.addEventListener("change", (e) => {
        // Only follow system if user hasn't set preference
        if (!localStorage.getItem("theme")) {
          document.documentElement.classList.toggle("dark", e.matches);
        }
      });
    </script>
  </head>
  <body
    data-sveltekit-preload-data="hover"
    class="bg-white text-black dark:bg-gray-900 dark:text-white"
  >
    <div style="display: contents">%sveltekit.body%</div>
  </body>
</html>

This short HTML code does a lot of things. First, the theme-color meta tags adapt the browser's UI (e.g., address bar) to match the theme. prefers-color-scheme is used to set different colors for light and dark modes. Then, in the script, we first checks if the user has a saved theme preference in localStorage. If not, it checks the OS-level dark mode setting using window.matchMedia('(prefers-color-scheme: dark)'). It then applies the appropriate theme by adding or removing the dark class to the <html> element. This class is used to toggle CSS styles (e.g., using dark:bg-gray-900 in the <body> class). Finally, it listens for changes in the OS-level dark mode setting and updates the theme accordingly (but only if the user hasn't explicitly set a preference). We also used the <body> element to set the background and text colors based on the presence of the dark class using Tailwind CSS classes.

Step 4: Theme switching logic with icons

Before we proceed to creating the authentication/login page, let's make a simple ThemeSwitcher.svelte component:

<script lang="ts">
  import Moon from "$lib/components/icons/Moon.svelte";
  import Sun from "$lib/components/icons/Sun.svelte";

  let { ...props } = $props();

  let isDark = $state(false);

  $effect(() => {
    isDark = document.documentElement.classList.contains("dark");
  });

  function toggleTheme() {
    isDark = !isDark;
    localStorage.setItem("theme", isDark ? "dark" : "light");
    document.documentElement.classList.toggle("dark", isDark);
  }
</script>

<button onclick="{toggleTheme}" {...props}>
  {#if isDark}
  <Sun />
  {:else}
  <Moon />
  {/if}
</button>

Since this app will be fully powered by Svelte 5, we are using the $props() rune to accept any attributes passed to the component as props, and the spread operator helps expand these attributes. We also declared a reactive variable with the $state rune, and it gets updated in the $effect rune and in the toggleTheme function. The $effect rune runs once on component initialization and whenever its dependencies change. In this case, it checks if the dark class is present on the <html> element and updates the isDark state accordingly. This ensures the component's initial state matches the current theme. As for the toggleTheme function, it gets called when the button is clicked and toggles the isDark state, saves the selected theme ("dark" or "light") to localStorage, and toggles the dark class on the <html> element. The <button> element calls the toggleTheme function on click and passes any additional props to the button. Inside the button, the {#if isDark}...{:else}...{/if} block conditionally renders either the Sun or Moon component based on the isDark state.

Note: Events are properties in Svelte 5

The on: directive was used in Svelte 4 to attach events to HTML elements. However, Svelte 5 changed that narrative by being more natural with HTML, which sees events as simply properties.

The sun and moon icons are simply svgs that have become svelte components:

<script lang="ts">
  let { ...props } = $props();
</script>

<svg
  class="h-6 w-6 text-yellow-400"
  fill="none"
  viewBox="0 0 24 24"
  stroke="currentColor"
  {...props}
>
  <path
    stroke-linecap="round"
    stroke-linejoin="round"
    stroke-width="2"
    d="M12 3v1m0 16v1m9-9h-1M4 12H3m15.364 6.364l-.707-.707M6.343 6.343l-.707-.707m12.728 0l-.707.707M6.343 17.657l-.707.707M16 12a4 4 0 11-8 0 4 4 0 018 0z"
  />
</svg>

<script lang="ts">
  let { ...props } = $props();
</script>

<svg
  class="h-6 w-6 text-gray-700 dark:text-gray-200"
  fill="none"
  viewBox="0 0 24 24"
  stroke="currentColor"
  {...props}
>
  <path
    stroke-linecap="round"
    stroke-linejoin="round"
    stroke-width="2"
    d="M20.354 15.354A9 9 0 018.646 3.646 9.003 9.003 0 0012 21a9.003 9.003 0 008.354-5.646z"
  />
</svg>

This is a nifty way to make svgs flexible as you can pass styles are other attributes and they will be reflected.

Step 5: Authentication or login page

Now, let's see what the login page will look like:

<script lang="ts">
    import { page } from '$app/state';
    import AiNode from '$lib/components/icons/AINode.svelte';
    import Calculator from '$lib/components/icons/Calculator.svelte';
    import FinChart from '$lib/components/icons/FinChart.svelte';
    import GitHub from '$lib/components/icons/GitHub.svelte';
    import Google from '$lib/components/icons/Google.svelte';
    import Logo from '$lib/components/logos/Logo.svelte';
    import ThemeSwitcher from '$lib/components/reusables/ThemeSwitcher.svelte';
    import { BASE_API_URI } from '$lib/utils/contants';
    import { fade } from 'svelte/transition';

    const next = page.url.searchParams.get('next') || '/';
</script>

<div
    class="relative min-h-screen bg-linear-to-br from-gray-100 to-gray-200 transition-colors duration-300 dark:from-gray-900 dark:to-gray-800"
>
    <!-- Theme Toggle -->
    <ThemeSwitcher
        class="dark:ring-black-500/50 absolute top-4 right-4 z-50 cursor-pointer rounded-full bg-white p-2 shadow-lg transition-all duration-300 hover:shadow-xl dark:bg-gray-700 dark:ring-2"
    />

    <!-- Decorative background elements -->
    <div class="absolute inset-0 z-0 overflow-hidden">
        <!-- AI Network Nodes -->
        <div class="floating-icons absolute top-10 left-10 opacity-20 dark:opacity-30">
            <AiNode />
        </div>
        <!-- Financial Chart -->
        <div class="floating-icons absolute right-20 bottom-32 opacity-20 dark:opacity-30">
            <FinChart />
        </div>
        <!-- Calculator Icon -->
        <div class="floating-icons absolute top-20 right-10 opacity-20 dark:opacity-30">
            <Calculator />
        </div>
    </div>

    <!-- Main content -->
    <div class="relative z-10 flex min-h-screen items-center justify-center">
        <div
            in:fade={{ duration: 300 }}
            class="w-full max-w-md space-y-8 rounded-xl bg-white/80 p-8 shadow-lg backdrop-blur-xs transition-all duration-300 dark:bg-gray-800/90 dark:shadow-gray-900/30"
        >
            <!-- Logo -->
            <div class="logo-container flex justify-center">
                <Logo isSmall={false} class="h-12 w-auto" />
            </div>
            <!-- Header -->
            <div class="text-center">
                <h2 class="mt-6 text-3xl font-extrabold text-gray-900 dark:text-white">Welcome back</h2>
                <p class="mt-2 text-sm text-gray-600 dark:text-gray-400">
                    Sign in to continue to your account
                </p>
                {#if page.url.searchParams.get('error')}
                    <p class="mt-2 text-sm text-red-500 dark:text-red-400">
                        Log in failed. Please try again.
                    </p>
                {/if}
            </div>

            <!-- Social Login Buttons -->
            <div class="mt-8 space-y-4">
                <!-- GitHub Login Button -->
                <a
                    href={`${BASE_API_URI}/v1/auth/github?next=${next}`}
                    class="flex w-full items-center justify-center rounded-md border border-gray-300 bg-white px-4 py-3 text-sm font-medium text-gray-700 shadow-xs transition-all duration-300 hover:bg-gray-50 dark:border-gray-600 dark:bg-gray-700 dark:text-gray-300 dark:hover:bg-gray-600"
                >
                    <GitHub />
                    Continue with GitHub
                </a>
                <!-- Google Login Button -->
                <a
                    href="/auth/google"
                    class="pointer-events-none flex w-full items-center justify-center rounded-md border border-gray-300 bg-white px-4 py-3 text-sm font-medium text-gray-700 opacity-50 shadow-xs transition-all duration-300 hover:bg-gray-50 dark:border-gray-600 dark:bg-gray-700 dark:text-gray-300 dark:hover:bg-gray-600"
                    aria-disabled="true"
                >
                    <Google />
                    Continue with Google
                </a>
            </div>

            <!-- Divider -->
            <div class="mt-6">
                <div class="relative">
                    <div class="absolute inset-0 flex items-center">
                        <div class="w-full border-t border-gray-300 dark:border-gray-600"></div>
                    </div>
                    <div class="relative flex justify-center text-sm">
                        <span class="bg-white px-2 text-gray-500 dark:bg-gray-800 dark:text-gray-400">
                            Policy Agreement
                        </span>
                    </div>
                </div>
            </div>
            <!-- Additional Info -->
            <div class="mt-6 text-center text-sm">
                <p class="text-gray-600 dark:text-gray-400">
                    By continuing, you agree to our
                    <a
                        href="/terms"
                        class="font-medium text-indigo-600 transition-colors duration-300 hover:text-indigo-500 dark:text-indigo-400 dark:hover:text-indigo-300"
                    >
                        Terms of Service
                    </a>
                    and
                    <a
                        href="/privacy"
                        class="font-medium text-indigo-600 transition-colors duration-300 hover:text-indigo-500 dark:text-indigo-400 dark:hover:text-indigo-300"
                    >
                        Privacy Policy
                    </a>
                </p>
            </div>
        </div>
    </div>
</div>

<style>
    .floating-icons {
        animation: float 6s ease-in-out infinite;
    }
    .floating-icons:nth-child(2) {
        animation-delay: 2s;
    }
    .floating-icons:nth-child(3) {
        animation-delay: 4s;
    }
    @keyframes float {
        0% {
            transform: translateY(0px) rotate(0deg);
        }
        50% {
            transform: translateY(-20px) rotate(360deg);
        }
        100% {
            transform: translateY(0px) rotate(0deg);
        }
    }
</style>

Though this code seems long, the only really important part is:

<script lang="ts">
    import { page } from '$app/state';
    ...
    import { BASE_API_URI } from '$lib/utils/contants';
    ...

    const next = page.url.searchParams.get('next') || '/';
</script>
...
<!-- Social Login Buttons -->
<div class="mt-8 space-y-4">
    <!-- GitHub Login Button -->
    <a
        href={`${BASE_API_URI}/v1/auth/github?next=${next}`}
        class="flex w-full items-center justify-center rounded-md border border-gray-300 bg-white px-4 py-3 text-sm font-medium text-gray-700 shadow-xs transition-all duration-300 hover:bg-gray-50 dark:border-gray-600 dark:bg-gray-700 dark:text-gray-300 dark:hover:bg-gray-600"
    >
        <GitHub />
        Continue with GitHub
    </a>
    <!-- Google Login Button -->
    <a
        href="/auth/google"
        class="pointer-events-none flex w-full items-center justify-center rounded-md border border-gray-300 bg-white px-4 py-3 text-sm font-medium text-gray-700 opacity-50 shadow-xs transition-all duration-300 hover:bg-gray-50 dark:border-gray-600 dark:bg-gray-700 dark:text-gray-300 dark:hover:bg-gray-600"
        aria-disabled="true"
    >
        <Google />
        Continue with Google
    </a>
</div>
...

These are just <a> elements whose URLs link directly to the API endpoints (Google yet to be implemented). The BASE_API_URI is exported by src/lib/utils/constants.ts:

export const BASE_API_URI = import.meta.env.DEV
  ? import.meta.env.VITE_BASE_API_URI_DEV
  : import.meta.env.VITE_BASE_API_URI_PROD;

It necessitates that you set either VITE_BASE_API_URI_DEV or VITE_BASE_API_URI_PRODas environment variables pointing to your server's base URL. In development, I set VITE_BASE_API_URI_DEV=http://localhost:3030/api in a .env file at the root of my SvelteKit project.

Tip: Environment Variables in Production

When deploying to a production server, configure the environment variable VITE_BASE_API_URI_PROD to point to your backend's production URL, including the /api path (e.g., https://your-production-url.com/api). Most deployment platforms offer a way to set these variables.

Back in our +page.svelte, we also retrieve the next page, which specifies where a user should be redirected after successful authentication. This is configured on a per-route basis. In addition to these core features, the page includes visual elements such as infinitely animating floating icons. These icons—AiNode (AI integration), FinChart (finance), and Calculator (analysis)—represent the application's key themes. Custom styles are applied to create a continuous vertical translation from 0px to -20px and back, combined with a simultaneous rotation from 0deg to 360deg. We could have used tailwind styles directly.

This article is getting long, so we will defer the user dashboard to the next article. However, there's a little safekeeping that needs to be done. Authenticated users should not be allowed to access this login page since they're already authenticated. To achieve this, we'll have a +page.server.ts file whose aim is to redirect any authenticated user from coming here:

import { redirect } from "@sveltejs/kit";
import type { PageServerLoad } from "./$types";

export const load: PageServerLoad = async ({ locals }) => {
  if (locals.user) {
    throw redirect(302, "/finanalyzer");
  }
  return {};
};

See you in the next one!

Outro

If you found this article valuable, consider sharing it with your network to help spread the knowledge!

Building a financial dashboard with HTML5, TailwindCSS v4, and Vanilla JavaScript

John Owolabi Idogun — Tue, 11 Feb 2025 17:00:00 +0000

Introduction

While developing the Financial data analyzer series, a dashboard became necessary for presenting the analyzed data intuitively and visually appealingly. Although I'm not a UI/UX designer, I appreciate well-designed interfaces with modern visual cues. This article details my process of creating such a dashboard from scratch with TailwindCSS v4 without using external frameworks.

Prerequisite

This article assumes you're familiar with HTML5, CSS3, and JavaScript (ES syntax). Familiarity with TailwindCSS is also helpful.

I'll present two ways to set up your development environment for working with TailwindCSS from scratch, without external frameworks.

Setup for Tailwind CSS v4 for Node.js users

To get started with Tailwind CSS v4 for your dashboard project using Node.js, follow these steps:

Create your project directory (e.g., finance-dashboard) and navigate into it:
```
mkdir finance-dashboard
cd finance-dashboard
```
Install Tailwind CSS, the Tailwind CSS CLI, and the @tailwindcss/forms plugin as development dependencies:
```
npm install -D tailwindcss @tailwindcss/cli @tailwindcss/forms
```

Create an input CSS file (e.g., assets/css/input.css) and add the following:

@import "tailwindcss";
@plugin "@tailwindcss/forms";

@custom-variant dark (&:where(.dark, .dark *));

@layer base {
  html {
    scroll-behavior: smooth;
  }

  body {
    overflow-y: scroll;
    font-family: "Fira Sans", sans-serif;
  }

  input[type],
  textarea,
  select {
    @apply appearance-none border-none ring-0 outline-hidden;

    &:focus {
      @apply border-none ring-0 outline-hidden;
    }

    &:focus-visible {
      @apply border-none ring-0 outline-hidden;
    }
  }

  button {
    @apply cursor-pointer;
  }
}

This CSS file imports Tailwind CSS, registers the @tailwindcss/forms plugin and sets up a custom dark variant for enabling dark mode via CSS classes. It also includes basic styling for smooth scrolling, font family, and form elements.

Note: TailwindCSS v4

We are using strictly TailwindCSS v4 here hence we ditched tailwind.config.[j|t]s file. You can read more in my v3 to v4 migration guide with plugins.

Create your main HTML file (e.g., index.html) with the following structure:

<!DOCTYPE html>
<html lang="en">
  <head>
    <meta charset="UTF-8" />
    <meta name="viewport" content="width=device-width, initial-scale=1.0" />
    <title>Dashboard | John Owolabi Idogun</title>
    <!-- Fonts -->
    <link rel="preconnect" href="https://fonts.googleapis.com" />
    <link rel="preconnect" href="https://fonts.gstatic.com" crossorigin />
    <link
      href="https://fonts.googleapis.com/css2?family=Fira+Code:wght@300..700&family=Fira+Sans:ital,wght@0,100;0,200;0,300;0,400;0,500;0,600;0,700;0,800;0,900;1,100;1,200;1,300;1,400;1,500;1,600;1,700;1,800;1,900&display=swap"
      rel="stylesheet"
    />
    <!-- ApexChart -->
    <script src="https://cdn.jsdelivr.net/npm/apexcharts"></script>
    <!-- Main Stylesheet -->
    <link rel="stylesheet" href="./assets/css/style.css" />
  </head>
  <body class="bg-white text-black dark:bg-gray-900 dark:text-white">
    <!-- body -->
    <script src="./assets/js/app.js"></script>
    <script src="./assets/js/index.charts.js"></script>
  </body>
</html>

This is a basic HTML structure that includes Google Fonts, ApexCharts (for placeholder charts), and links to your compiled CSS and JavaScript files. The body includes classes for light and dark modes.

Generate the compiled CSS file, assets/css/style.css:

npx tailwindcss -i ./assets/css/input.css -o ./assets/css/style.css --watch --minify

Tailwind CSS v4 Setup without Node.js

For those who prefer not to use Node.js, you can use TailwindCSS's Standalone CLI. Follow the guide to install it based on your operating system. Then, complete steps 1, 3, and 4 from the Node.js setup, skipping steps 2 and 5. To compile your CSS, run:

./tailwindcss -i input.css -o output.css --watch --minify

This setup provides a foundation for building the dashboard with Tailwind CSS v4, utilizing vanilla JavaScript for interactivity and ApexCharts for data visualization.

Source code

Sirneij / finance-dashboard

An aesthetic personal finance dashboard built with vanilla JS, tailwindcss v4 and HTML5

Finance Dashboard

A responsive financial dashboard built with HTML, Tailwind CSS v4, and vanilla JavaScript.

Features

Responsive Design: Adapts seamlessly between mobile and desktop views
Collapsible Sidebar: Full-width and compact viewing options
Dark Mode Support: Automatic system theme detection with manual toggle
Real-time Data Visualization: Using ApexCharts for financial data display
Mobile-First Approach: Optimized for all screen sizes

Project Structure

finance-dashboard/
'
├── README
├── assets
│   ├── css
│   │   ├── input.css
│   │   └── style.css
│   ├── images
│   │   ├── favicons
│   │   │   ├── apple-touch-icon.png
│   │   │   ├── favicon-96x96.png
│   │   │   ├── favicon.ico
│   │   │   ├── favicon.svg
│   │   │   ├── site.webmanifest
│   │   │   ├── web-app-manifest-192x192.png
│   │   │   └── web-app-manifest-512x512.png
│   │   ├── logo-small.svg
│   │   └── logo.svg
│   └── js
│       ├── app.js
│       └── index.charts.js
├── index.html
├── package-lock.json
├── package.json
└── pages
    ├── behavior.html
    └── transactions.html

Getting Started

…

View on GitHub

Implementation

Step 1: Header and Sidebar

First off, we will build out the header and sidebar of the dashboard. Let's add this to the body of the page:

<div
  class="relative h-screen overflow-hidden bg-gray-100 dark:bg-gray-900"
  id="main-content"
>
  <!-- Sidebar -->
  <aside
    id="sidebar"
    class="fixed inset-y-0 left-0 z-30 transform bg-white transition-all duration-300 dark:bg-gray-800"
  >
    <div class="flex h-16 items-center justify-between px-4">
      <a class="flex items-center" href="/">
        <img
          id="logo"
          src="./assets/images/logo.svg"
          alt="Logo"
          class="h-12 w-auto"
        />
      </a>
      <button
        onclick="sidebarController.toggle()"
        class="rounded-sm p-1 hover:bg-gray-100 dark:hover:bg-gray-700"
        aria-label="Toggle sidebar"
      >
        <svg
          class="h-6 w-6 text-gray-600 dark:text-gray-300"
          viewBox="0 0 24 24"
          fill="none"
        >
          <path
            id="toggle-path"
            stroke="currentColor"
            stroke-width="2"
            d="M15 19l-7-7 7-7"
          />
        </svg>
      </button>
    </div>

    <!-- Navigation items -->
    <nav
      id="sidebar-nav"
      class="flex h-[calc(100vh-4rem)] flex-col justify-between px-4"
    >
      <!-- Main Navigation -->
      <div class="space-y-2" id="main-nav">
        <a
          href="index.html"
          data-nav-link
          class="flex items-center rounded-lg px-4 py-2 text-gray-600 hover:bg-gray-100 dark:text-gray-300 dark:hover:bg-gray-700"
          title="Overview"
        >
          <svg
            class="h-6 w-6"
            fill="none"
            viewBox="0 0 24 24"
            stroke="currentColor"
          >
            <path
              stroke-linecap="round"
              stroke-linejoin="round"
              stroke-width="2"
              d="M3 12l2-2m0 0l7-7 7 7M5 10v10a1 1 0 001 1h3m10-11l2 2m-2-2v10a1 1 0 01-1 1h-3m-6 0a1 1 0 001-1v-4a1 1 0 011-1h2a1 1 0 011 1v4a1 1 0 001 1m-6 0h6"
            />
          </svg>
          <span class="ml-3" data-nav-label>Overview</span>
        </a>

        <a
          href="behavior.html"
          data-nav-link
          class="flex items-center rounded-lg px-4 py-2 text-gray-600 hover:bg-gray-100 dark:text-gray-300 dark:hover:bg-gray-700"
          title="Behavior Analysis"
        >
          <svg
            class="h-6 w-6"
            fill="none"
            viewBox="0 0 24 24"
            stroke="currentColor"
          >
            <path
              stroke-linecap="round"
              stroke-linejoin="round"
              stroke-width="2"
              d="M9 19v-6a2 2 0 00-2-2H5a2 2 0 00-2 2v6a2 2 0 002 2h2a2 2 0 002-2zm0 0V9a2 2 0 012-2h2a2 2 0 012 2v10m-6 0a2 2 0 002 2h2a2 2 0 002-2m0 0V5a2 2 0 012-2h2a2 2 0 012 2v14a2 2 0 01-2 2h-2a2 2 0 01-2-2z"
            />
          </svg>
          <span class="ml-3" data-nav-label>Behavior</span>
        </a>

        <a
          href="transactions.html"
          data-nav-link
          class="flex items-center rounded-lg px-4 py-2 text-gray-600 hover:bg-gray-100 dark:text-gray-300 dark:hover:bg-gray-700"
          title="Transaction History"
        >
          <svg
            class="h-6 w-6"
            fill="none"
            viewBox="0 0 24 24"
            stroke="currentColor"
          >
            <path
              stroke-linecap="round"
              stroke-linejoin="round"
              stroke-width="2"
              d="M4 8h16M4 16h16"
            />
          </svg>
          <span class="ml-3" data-nav-label>Transactions</span>
        </a>
        <a
          href="https://johnowolabiidogun.dev"
          data-nav-link
          class="flex items-center rounded-lg px-4 py-2 text-gray-600 hover:bg-gray-100 dark:text-gray-300 dark:hover:bg-gray-700"
          title="About Developer"
          target="_blank"
        >
          <svg
            class="h-6 w-6"
            fill="none"
            viewBox="0 0 24 24"
            stroke="currentColor"
          >
            <path
              stroke-linecap="round"
              stroke-linejoin="round"
              stroke-width="2"
              d="M13 10V3L4 14h7v7l9-11h-7z"
            />
          </svg>
          <span class="ml-3" data-nav-label>About Developer</span>
        </a>
      </div>

      <!-- Logout Section -->
      <div class="shrink-0">
        <div class="mb-2 h-px w-full bg-gray-200 dark:bg-gray-700"></div>
        <a
          href="#"
          class="group mb-4 flex items-center rounded-lg px-4 py-2 text-red-600 hover:bg-red-50 dark:text-red-400 dark:hover:bg-red-900/20"
          title="Logout from application"
        >
          <svg
            class="h-6 w-6"
            fill="none"
            viewBox="0 0 24 24"
            stroke="currentColor"
          >
            <path
              stroke-linecap="round"
              stroke-linejoin="round"
              stroke-width="2"
              d="M17 16l4-4m0 0l-4-4m4 4H7m6 4v1a3 3 0 01-3 3H6a3 3 0 01-3-3V7a3 3 0 013-3h4a3 3 0 013 3v1"
            />
          </svg>
          <span class="ml-3" data-nav-label>Logout</span>
        </a>
      </div>
    </nav>
  </aside>

  <!-- Main Content -->
</div>

The entire page is meant to take the height of the screen (h-screen). This ensures the page remains in view. The sidebar inherits this property. It contains both icons and labels. We'll use both so that when the sidebar is fully open, both are visible, but when it's collapsed, only the icons are displayed.

Next, we will add the main content markup:

<!-- Main Content -->
<div
  class="relative h-full transform transition-all duration-300 md:translate-x-0 md:ml-64"
  id="main"
>
  <header
    class="sticky top-0 z-10 flex h-16 items-center justify-between border-b border-gray-200 bg-white px-6 dark:border-gray-700 dark:bg-gray-800"
  >
    <div class="flex items-center gap-4">
      <!-- Mobile menu button -->
      <button
        class="rounded-lg p-2 text-gray-500 hover:bg-gray-100 md:hidden dark:text-gray-400 dark:hover:bg-gray-700"
        onclick="sidebarController.toggle()"
        aria-label="Toggle Menu"
        title="Toggle Menu"
      >
        <svg
          xmlns="http://www.w3.org/2000/svg"
          width="24"
          height="24"
          viewBox="0 0 24 24"
          fill="none"
          stroke="currentColor"
          stroke-width="2"
          stroke-linecap="round"
          stroke-linejoin="round"
          class="h-6 w-6"
        >
          <path stroke="none" d="M0 0h24v24H0z" fill="none" />
          <path d="M4 6h16" />
          <path d="M7 12h13" />
          <path d="M10 18h10" />
        </svg>
      </button>
      <h1 class="text-2xl font-semibold text-gray-800 dark:text-white">
        Dashboard
      </h1>
    </div>
    <button
      id="theme-switcher"
      onclick="themeController.toggle()"
      class="rounded-full bg-white p-2 shadow-lg transition-all duration-300 hover:shadow-xl dark:bg-gray-700 dark:ring-2"
      aria-label="Toggle theme"
    >
      <!-- Sun icon for dark mode -->
      <svg
        id="sun-icon"
        class="h-6 w-6 text-gray-600 dark:text-gray-300 hidden"
        fill="none"
        viewBox="0 0 24 24"
        stroke="currentColor"
      >
        <path
          stroke-linecap="round"
          stroke-linejoin="round"
          stroke-width="2"
          d="M12 3v1m0 16v1m9-9h-1M4 12H3m15.364 6.364l-.707-.707M6.343 6.343l-.707-.707m12.728 0l-.707.707M6.343 17.657l-.707.707M16 12a4 4 0 11-8 0 4 4 0 018 0z"
        />
      </svg>
      <!-- Moon icon for light mode -->
      <svg
        id="moon-icon"
        class="h-6 w-6 text-gray-600 dark:text-gray-300"
        fill="none"
        viewBox="0 0 24 24"
        stroke="currentColor"
      >
        <path
          stroke-linecap="round"
          stroke-linejoin="round"
          stroke-width="2"
          d="M20.354 15.354A9 9 0 018.646 3.646 9.003 9.003 0 0012 21a9.003 9.003 0 008.354-5.646z"
        />
      </svg>
    </button>
  </header>
  <main class="h-[calc(100vh-4rem)] overflow-y-auto p-6">
    <section class="space-y-6">
      <!-- Welcome Section -->
    </section>
  </main>
</div>

At the top, we have the header with the "Dashboard" inscription. It also houses the icons that toggle light/dark modes. The main page takes the remaining height available on the page (h-[calc(100vh-4rem)]) and allows vertical scrolling in case of overflow (overflow-y-auto).

The remaining markups are easy to follow, so we won't paste them here. You can always visit the project's repo to copy them. They look like this for now:

We'll proceed to write the theme-switching logic and responsive triggers.

Step 2: Responsive triggers and Theme Switching logic

Make your assets/js/app.js look like this:

class SidebarController {
  constructor() {
    this.isSidebarOpen = true;
    this.isMobile = window.innerWidth < 768;
    this.navLinks = document.querySelectorAll("[data-nav-link]");
    this.host = window.location.origin;

    // Cache DOM elements
    this.sidebar = document.getElementById("sidebar");
    this.main = document.getElementById("main");
    this.mobileOverlay = document.getElementById("mobile-overlay");
    this.logo = document.getElementById("logo");
    this.togglePath = document.getElementById("toggle-path");
    this.navLabels = document.querySelectorAll("[data-nav-label]");

    // Bind methods
    this.checkWidth = this.checkWidth.bind(this);
    this.toggle = this.toggle.bind(this);

    // Set initial state
    requestAnimationFrame(() => {
      this.checkWidth();
      this.updateUI(true); // true for initial load
      this.highlightCurrentPage();
    });

    window.addEventListener("resize", this.checkWidth);
  }

  checkWidth() {
    const wasMobile = this.isMobile;
    this.isMobile = window.innerWidth < 768;

    if (wasMobile !== this.isMobile) {
      this.isSidebarOpen = !this.isMobile;
      this.updateUI();
    }
  }

  toggle() {
    this.isSidebarOpen = !this.isSidebarOpen;
    this.updateUI();
  }

  updateUI(isInitialLoad = false) {
    // Mobile specific
    if (this.isMobile) {
      this.sidebar.classList.toggle("-translate-x-full", !this.isSidebarOpen);
      this.main.classList.toggle("overflow-hidden", this.isSidebarOpen);
      this.mobileOverlay?.classList.toggle("hidden", !this.isSidebarOpen);

      // Reset desktop classes
      this.main.classList.remove("md:ml-64", "md:ml-20");
    } else {
      // Desktop specific
      if (isInitialLoad) {
        // Force initial margin on load
        this.main.classList.add("md:ml-64");
      } else {
        this.main.classList.toggle("md:ml-64", this.isSidebarOpen);
        this.main.classList.toggle("md:ml-20", !this.isSidebarOpen);
      }

      // Reset mobile classes
      this.sidebar.classList.remove("-translate-x-full");
      this.main.classList.remove("overflow-hidden");
      this.mobileOverlay?.classList.add("hidden");
    }

    // Common updates
    this.sidebar.classList.toggle("w-64", this.isSidebarOpen);
    this.sidebar.classList.toggle("w-20", !this.isSidebarOpen);

    // Update logo
    if (this.logo) {
      const logoURL = this.host.includes("sirneij.github.io")
        ? `${this.host}/finance-dashboard/assets/images/logo.svg`
        : "./assets/images/logo.svg";
      const logoSmallURL = this.host.includes("sirneij.github.io")
        ? `${this.host}/finance-dashboard/assets/images/logo-small.svg`
        : "./assets/images/logo-small.svg";

      this.logo.src = this.isSidebarOpen
        ? logoURL.replace("null", ".")
        : logoSmallURL.replace("null", ".");

      this.logo.classList.toggle("h-12", this.isSidebarOpen);
      this.logo.classList.toggle("h-8", !this.isSidebarOpen);
    }

    // Update toggle icon
    if (this.togglePath) {
      this.togglePath.setAttribute(
        "d",
        this.isSidebarOpen ? "M15 19l-7-7 7-7" : "M9 19l7-7-7-7"
      );
    }

    // Update labels
    this.navLabels.forEach((label) => {
      label.style.display = this.isSidebarOpen ? "block" : "none";
    });
  }
  highlightCurrentPage() {
    const currentPath = window.location.pathname;

    this.navLinks.forEach((link) => {
      const linkPath = link.getAttribute("href");

      const isActive =
        currentPath === linkPath ||
        (currentPath === "/" && linkPath === "/") ||
        (currentPath !== "/" &&
          linkPath !== "/" &&
          currentPath.includes(linkPath)) ||
        (currentPath === "/" && linkPath === "index.html");

      // Remove existing active classes
      link.classList.remove(
        "bg-gray-100",
        "text-primary-600",
        "dark:bg-gray-700",
        "dark:text-primary-500"
      );

      // Add active classes if current page
      if (isActive) {
        link.classList.add(
          "bg-gray-100",
          "text-primary-600",
          "dark:bg-gray-700",
          "dark:text-primary-500"
        );
      }
    });
  }
}

// Theme handling
const themeController = {
  init() {
    const userTheme = localStorage.getItem("theme");
    const systemTheme = window.matchMedia("(prefers-color-scheme: dark)");
    const theme = userTheme || (systemTheme.matches ? "dark" : "light");

    this.updateTheme(theme === "dark");
    systemTheme.addEventListener("change", (e) => {
      if (!localStorage.getItem("theme")) {
        this.updateTheme(e.matches);
      }
    });
  },

  toggle() {
    const isDark = document.documentElement.classList.contains("dark");
    this.updateTheme(!isDark);
    localStorage.setItem("theme", !isDark ? "dark" : "light");
  },

  updateTheme(isDark) {
    document.documentElement.classList.toggle("dark", isDark);
    document.getElementById("sun-icon").classList.toggle("hidden", !isDark);
    document.getElementById("moon-icon").classList.toggle("hidden", isDark);
  },
};

// Initialize when DOM is ready
document.addEventListener("DOMContentLoaded", () => {
  window.sidebarController = new SidebarController();
  themeController.init();
});

This JavaScript code sets up the responsive sidebar and theme-switching logic for the dashboard. SidebarController handles the sidebar's behavior on different screen sizes, including toggling the sidebar and highlighting the current page in the navigation. The ThemeController manages the theme (light/dark) based on user preference or system settings, persisting the choice in local storage. The code uses classes and event listeners for efficient state management and dynamic UI updates.

I opted for classes due to state management issues, especially isSidebarOpen and isMobile. Binding them up here makes it very easy to elegantly manage.

The collapsed versions should look like these:

Step 3: ApexCharts configurations

For a little taste of the awesome and relatively lightweight charting library (the reason it was preferred compared to Chart.js):

function initializeMonthlyChart() {
  const options = {
    series: [
      {
        name: "Income",
        data: [3000, 3500, 4000, 3800, 4200, 4500],
      },
      {
        name: "Expenses",
        data: [2500, 2800, 3000, 2900, 3100, 3300],
      },
      {
        name: "Savings",
        data: [500, 700, 1000, 900, 1100, 1200],
      },
    ],
    chart: {
      type: "area",
      height: 300,
      toolbar: {
        show: true,
        tools: {
          download: true,
          selection: true,
          zoom: true,
          zoomin: true,
          zoomout: true,
          pan: true,
          reset: true,
        },
        autoSelected: "zoom",
      },
      fontFamily: "inherit",
      background: "transparent",
    },
    colors: ["#22c55e", "#ef4444", "#3b82f6"],
    fill: {
      type: "gradient",
      gradient: {
        shadeIntensity: 1,
        inverseColors: false,
        opacityFrom: 0.5,
        opacityTo: 0,
        stops: [0, 90, 100],
      },
    },
    dataLabels: { enabled: false },
    stroke: {
      width: 2,
      curve: "smooth",
    },
    grid: {
      borderColor: "rgba(156, 163, 175, 0.1)",
      strokeDashArray: 4,
      yaxis: { lines: { show: true } },
      xaxis: { lines: { show: false } },
    },
    xaxis: {
      categories: ["Jan", "Feb", "Mar", "Apr", "May", "Jun"],
      labels: {
        style: {
          colors: "rgba(156, 163, 175, 0.9)",
        },
      },
    },
    yaxis: {
      labels: {
        formatter: (value) => `$${value}`,
        style: {
          colors: "rgba(156, 163, 175, 0.9)",
        },
      },
    },
    legend: {
      show: false,
    },
    theme: {
      mode: document.documentElement.classList.contains("dark")
        ? "dark"
        : "light",
    },
  };

  return new ApexCharts(
    document.querySelector("#monthly-summary-chart"),
    options
  );
}

function initializeFinancialChart() {
  const options = {
    series: [
      {
        name: "Income",
        data: [1500, 2000, 1800, 2200, 1900],
      },
      {
        name: "Expenses",
        data: [1200, 1400, 1100, 1600, 1300],
      },
      {
        name: "Balance",
        data: [300, 600, 700, 600, 600],
      },
    ],
    chart: {
      type: "area",
      height: "100%",
      toolbar: {
        show: true,
        tools: {
          download: true,
          selection: true,
          zoom: true,
          zoomin: true,
          zoomout: true,
          pan: true,
          reset: true,
        },
        autoSelected: "zoom",
      },
      fontFamily: "inherit",
      background: "transparent",
    },
    colors: ["#22c55e", "#ef4444", "#3b82f6"],
    fill: {
      type: "gradient",
      gradient: {
        shadeIntensity: 1,
        inverseColors: false,
        opacityFrom: 0.5,
        opacityTo: 0,
        stops: [0, 90, 100],
      },
    },
    dataLabels: { enabled: false },
    stroke: {
      width: 2,
      curve: "smooth",
      dashArray: [0, 0, 5],
    },
    grid: {
      borderColor: "rgba(156, 163, 175, 0.1)",
      strokeDashArray: 4,
      yaxis: { lines: { show: true } },
      xaxis: { lines: { show: false } },
    },
    xaxis: {
      categories: ["Day 1", "Day 2", "Day 3", "Day 4", "Day 5"],
      labels: {
        style: {
          colors: "rgba(156, 163, 175, 0.9)",
        },
      },
    },
    yaxis: {
      labels: {
        formatter: (value) => `$${value}`,
        style: {
          colors: "rgba(156, 163, 175, 0.9)",
        },
      },
    },
    legend: {
      show: false,
    },
    theme: {
      mode: document.documentElement.classList.contains("dark")
        ? "dark"
        : "light",
    },
  };

  return new ApexCharts(
    document.querySelector("#financial-trends-chart"),
    options
  );
}

// Initialize charts when DOM is loaded
document.addEventListener("DOMContentLoaded", () => {
  const monthlyChart = initializeMonthlyChart();
  const financialChart = initializeFinancialChart();

  monthlyChart.render();
  financialChart.render();

  // Handle theme changes
  const observer = new MutationObserver(() => {
    const isDark = document.documentElement.classList.contains("dark");
    monthlyChart.updateOptions({ theme: { mode: isDark ? "dark" : "light" } });
    financialChart.updateOptions({
      theme: { mode: isDark ? "dark" : "light" },
    });
  });

  observer.observe(document.documentElement, {
    attributes: true,
    attributeFilter: ["class"],
  });

  // Handle resize
  window.addEventListener(
    "resize",
    debounce(() => {
      monthlyChart.updateOptions({});
      financialChart.updateOptions({});
    }, 300)
  );
});

function debounce(fn, ms) {
  let timer;
  return function () {
    clearTimeout(timer);
    timer = setTimeout(() => fn.apply(this, arguments), ms);
  };
}

This code configures and initializes ApexCharts for the dashboard. initializeMonthlyChart and initializeFinancialChart functions define the options for two different area charts, including series data, chart type, colors, and grid settings. The code also includes a MutationObserver to handle theme changes and a debounce function to optimize resize event handling, ensuring the charts are responsive and adapt to the selected theme.

There are other pages implemented and the repo has them. You can also preview its live version.

Outro

If you found this article valuable, consider sharing it with your network to help spread the knowledge!

Consuming paginated API using periodic Celery task in a Django Application

John Owolabi Idogun — Sat, 08 Feb 2025 20:22:23 +0000

Introduction

API (Application Programming Interface) consumption is the new normal in contemporary times as many software products have shifted focus on decoupling Backend and Frontend codebases. Backend Engineers are tasked with writing consumable APIs that their Frontend counterparts consume. In some cases, even Backend Engineers utilize some other API services to accomplish their tasks.

Some services provide an enormously large dataset so making them accessible at a single API call might not be great. Pagination then comes to the rescue. Many APIs are now paginated to make available a fraction of the data. To access other fractions, you need some extra tasks.

This article demonstrates how to set up Celery background tasks to consume paginated APIs periodically. We'll explore iterative and recursive approaches for APIs paginated using page parameters and those using next URLs. The fetched data will be stored in a Django model, overwriting previous data. Note that persisting historical data is outside the scope of this article but will be addressed in a future post on building a data warehouse.

Prerequisite

Basic familiarity with Django is assumed. Refer to the Django tutorial for an introduction.

Source code

Sirneij / django_excel

Exporting Django model data as excel file (.xlsx) using openpyxl library and Google Spreadsheet API

django_excel

This repository accompanies this tutorial on dev.to. It has been deployed to Heroku and can be accessed live via this link.

NOTE: If you use Coingecko's API, when you use my code, `CGSIRNEIJ`, I get some commissions. That can be a good way to help me.

Run Locally

To run the project locally:

Create a virtual environment using venv, poetry, virtualenv, or pipenv. I recommend virtualenv.
Activate the virtual environment.
Install dependencies:
```
pip install -r requirements.txt
```
Migrate the database:
```
python manage.py migrate
```
Run the project:
```
python manage.py runserver
```
In another terminal, issue this command to start celery:
```
celery -A django_excel worker -l info -B
```

Run Tests Locally

To run the tests:

pytest --nomigrations --reuse-db -W error::RuntimeWarning --cov=core --cov-report=html tests/

View on GitHub

Implementation

Step 1: Setup Django project and app with celery

Create a Django project named django_excel within a virtual environment. Ensure that django and celery are installed in your environment.

virtualenv➜  django_excel git:(main) django-admin startproject django_excel .

Create an app named core:

virtualenv➜  django_excel git:(main) python manage.py startapp core

...
# Application definition

INSTALLED_APPS = [
    'django.contrib.admin',
    'django.contrib.auth',
    'django.contrib.contenttypes',
    'django.contrib.sessions',
    'django.contrib.messages',
    'django.contrib.staticfiles',

    'core.apps.CryptoappConfig', #Our new app
]
...

It is time to set up our application to utilize Celery. To do this, create a file aptly named celery.py in your project's directory and paste the following snippet:

import os

from celery import Celery

# Set the default Django settings module for the 'celery' program.
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'django_excel.settings')

app = Celery('django_excel')

# Using a string here means the worker doesn't have to serialize
# the configuration object to child processes.
# - namespace='CELERY' means all celery-related configuration keys
#   should have a `CELERY_` prefix.
app.config_from_object('django.conf:settings', namespace='CELERY')

# Load task modules from all registered Django apps.
app.autodiscover_tasks()


@app.task(bind=True)
def debug_task(self):
    print(f'Request: {self.request!r}')

That was directly lifted from Celery Django documentation. Ensure you modify lines 6 and 8 to reflect your project's name. The namespace in line 14 enables you to prefix all celery-related configurations in your settings.py file with CELERY such as CELERY_BROKER_URL.

Note: Capitalization of Celery-Related Configurations

Because you are literarily providing constants, the celery-related configurations in your settings.py file are capitalized. For instance, one of the configurations is beat_schedule which in Django, becomes CELERY_BEAT_SCHEDULE.

Next, open open your project's __init__.py and append the following:

# This will make sure the app is always imported when
# Django starts so that shared_task will use this app.
from .celery import app as celery_app

__all__ = ('celery_app',)

To conclude celery-related configurations, let's set the following in settings.py:

...
CELERY_BROKER_URL: str = config('REDIS_URL', default='redis://localhost:6379/0')
CELERY_RESULT_BACKEND: str = config('REDIS_URL', default='redis://localhost:6379/0')
CELERY_ACCEPT_CONTENT: list[str] = ['application/json']
CELERY_TASK_SERIALIZER: str = 'json'
CELERY_RESULT_SERIALIZER: str = 'json'

We are using redis as our broker. You can opt for RabbitMQ which is supported out-of-box by celery.

In the settings above, I am linking CELERY_BROKER_URL to an environment variable named REDIS_URL. It normally should look like redis://127.0.0.1:6379 on a Linux system. That means I could have set my CELERY_BROKER_URL and CELERY_RESULT_BACKEND as:

...
from celery.schedules import crontab
from decouple import Csv, config
...
CELERY_BROKER_URL: str = config('REDIS_URL', default='redis://localhost:6379/0')
CELERY_RESULT_BACKEND: str = config('REDIS_URL', default='redis://localhost:6379/0')

CELERY_ACCEPT_CONTENT: list[str] = ['application/json']
CELERY_TASK_SERIALIZER: str = 'json'
CELERY_RESULT_SERIALIZER: str = 'json'

Note that CELERY_RESULT_BACKEND is optional as well as CELERY_ACCEPT_CONTENT, CELERY_TASK_SERIALIZER, and CELERY_RESULT_SERIALIZER. However, not setting the last three might result in some runtime errors mostly when dealing with databases in asynchronous email broadcasting with celery.

The stage is now set, let's set up our database. We will be consuming CoinGecko's API and will be saving some data.

Step 2: Define `FullCoin` model

Our model will look like this:

from django.db import models

class FullCoin(models.Model):
    coin_id = models.CharField(max_length=100, primary_key=True)
    symbol = models.CharField(max_length=10, null=True, blank=True)
    name = models.CharField(max_length=100, null=True, blank=True)
    image = models.URLField(null=True, blank=True)
    current_price = models.DecimalField(max_digits=20, decimal_places=2, null=True, blank=True)
    market_cap = models.BigIntegerField(null=True, blank=True)
    market_cap_rank = models.IntegerField(null=True, blank=True)
    fully_diluted_valuation = models.BigIntegerField(null=True, blank=True)
    total_volume = models.BigIntegerField(null=True, blank=True)
    high_24h = models.DecimalField(max_digits=20, decimal_places=2, null=True, blank=True)
    low_24h = models.DecimalField(max_digits=20, decimal_places=2, null=True, blank=True)
    price_change_24h = models.DecimalField(max_digits=20, decimal_places=2, null=True, blank=True)
    price_change_percentage_24h = models.DecimalField(max_digits=10, decimal_places=5, null=True, blank=True)
    market_cap_change_24h = models.BigIntegerField(null=True, blank=True)
    market_cap_change_percentage_24h = models.DecimalField(max_digits=10, decimal_places=5, null=True, blank=True)
    circulating_supply = models.DecimalField(max_digits=20, decimal_places=2, null=True, blank=True)
    total_supply = models.DecimalField(max_digits=20, decimal_places=2, null=True)
    max_supply = models.DecimalField(max_digits=20, decimal_places=2, null=True)
    ath = models.DecimalField(max_digits=20, decimal_places=2, null=True, blank=True)
    ath_change_percentage = models.DecimalField(max_digits=10, decimal_places=5, null=True, blank=True)
    ath_date = models.DateTimeField(null=True, blank=True)
    atl = models.DecimalField(max_digits=20, decimal_places=2, null=True, blank=True)
    atl_change_percentage = models.DecimalField(max_digits=20, decimal_places=5, null=True, blank=True)
    atl_date = models.DateTimeField(null=True, blank=True)
    last_updated = models.DateTimeField(null=True, blank=True)

    def __str__(self):
        return f"{self.name} ({self.symbol.upper()})"

These are all the fields taken directly from CoinGecko's public API for coin markets:

[
  {
    "id": "bitcoin",
    "symbol": "btc",
    "name": "Bitcoin",
    "image": "https://assets.coingecko.com/coins/images/1/large/bitcoin.png?1696501400",
    "current_price": 70187,
    "market_cap": 1381651251183,
    "market_cap_rank": 1,
    "fully_diluted_valuation": 1474623675796,
    "total_volume": 20154184933,
    "high_24h": 70215,
    "low_24h": 68060,
    "price_change_24h": 2126.88,
    "price_change_percentage_24h": 3.12502,
    "market_cap_change_24h": 44287678051,
    "market_cap_change_percentage_24h": 3.31157,
    "circulating_supply": 19675987,
    "total_supply": 21000000,
    "max_supply": 21000000,
    "ath": 73738,
    "ath_change_percentage": -4.77063,
    "ath_date": "2024-03-14T07:10:36.635Z",
    "atl": 67.81,
    "atl_change_percentage": 103455.83335,
    "atl_date": "2013-07-06T00:00:00.000Z",
    "roi": null,
    "last_updated": "2024-04-07T16:49:31.736Z"
  }
]

Note:Commission from referral

If you use Coingecko's API, when you use my code, CGSIRNEIJ, I get some commissions. That can be a good way to support me.

null=True makes a column nullable in SQL:

CREATE TABLE IF NOT EXISTS full_coin(
    symbol VARCHAR(10) NULL,
    ...
)

null=False or leaving it unset makes the column non-nullable:

CREATE TABLE IF NOT EXISTS full_coin(
    symbol VARCHAR(10) NOT NULL,
    ...
)

blank=True allows the field to be optional in forms and the admin page.

Talking about the admin site, let's register the model:

from django.contrib import admin

from core.models import FullCoin

@admin.register(FullCoin)
class FullCoinAdmin(admin.ModelAdmin):
    list_display = (
        'coin_id',
        'symbol',
        'name',
        'current_price',
        'market_cap',
        'market_cap_rank',
        'fully_diluted_valuation',
        'total_volume',
        'high_24h',
        'low_24h',
        'price_change_24h',
        'price_change_percentage_24h',
        'market_cap_change_24h',
        'market_cap_change_percentage_24h',
        'circulating_supply',
        'total_supply',
        'max_supply',
        'ath',
        'ath_change_percentage',
        'ath_date',
        'atl',
        'atl_change_percentage',
        'atl_date',
        'last_updated',
    )

With this, you can migrate your database:

```bash :Terminal:
virtualenv➜ django_excel git:(main) python manage.py makemigrations # create migration file(s)
virtualenv➜ django_excel git:(main) python manage.py migrate # create the tables in the db




Optionally, you can create a superuser:



```bash :Terminal:
virtualenv➜  django_excel git:(main) python manage.py createsuperuser

Follow the prompts.

Step 3: Create and Register Periodic Tasks

Here is the juicy part:

...
import logging
import time

import requests
from celery import shared_task
from django.conf import settings

from core.models import FullCoin

logger = logging.getLogger(__name__)


def build_api_url(page: int) -> str:
    """Build the API URL."""
    market_currency_order = 'markets?vs_currency=usd&order=market_cap_desc&'
    per_page = f'per_page=50&page={page}&sparkline=false'
    return f'{settings.BASE_API_URL}/coins/{market_currency_order}{per_page}'


def fetch_coins_iteratively() -> Generator[dict, None, None]:
    """Fetch coins data from API using generator."""
    page = 1
    while True:
        try:
            url = build_api_url(page)
            response = requests.get(url)
            coin_data = response.json()

            # Check for rate limit response
            if isinstance(coin_data, dict) and coin_data.get('status', {}).get('error_code') == 429:
                logger.warning("Rate limit exceeded. Waiting 60 seconds...")
                time.sleep(60)
                continue

            # Check for empty response (end of pagination)
            if not coin_data:
                break

            yield from coin_data
            logger.info(f"Fetched page {page} with {len(coin_data)} coins")
            page += 1

            time.sleep(1)  # Be nice to the API

        except requests.exceptions.RequestException as e:
            logger.error(f"Request failed on page {page}: {e}")
            raise


@shared_task(
    bind=True,
    autoretry_for=(Exception,),
    retry_backoff=True,
    retry_backoff_max=600,
    max_retries=5,
)
def get_full_coin_data_iteratively_for_page(self) -> None:
    """Get full coin data iteratively for each page."""
    try:
        # Use list comprehension to collect coins in batches
        batch_size = 100
        coins_batch = []

        for coin in fetch_coins_iteratively():
            coins_batch.append(coin)

            if len(coins_batch) >= batch_size:
                logger.info(f"Processing batch of {len(coins_batch)} coins")
                store_data(coins_batch)
                coins_batch = []

        # Process remaining coins
        if coins_batch:
            logger.info(f"Processing final batch of {len(coins_batch)} coins")
            store_data(coins_batch)

    except Exception as e:
        logger.error(f"Failed to process coins: {e}")
        raise self.retry(exc=e)

The build_api_url helps continuously build CoinGecko API url based on the supplied page number. The BASE_API_URL is:

```python :django_excel/settings.py:
...
BASE_API_URL: str = 'https://api.coingecko.com/api/v3'




`fetch_coins_iteratively` is the core of the program. It starts with the first page and does an "infinite" loop which breaks only when there's no data returned by the API using the iterative strategy.

Its recursive alternative is:



```python
def fetch_coins_recursively(page: int = 1) -> Generator[dict, None, None]:
    """Fetch coins data from API recursively using generator."""
    try:
        url = build_api_url(page)
        response = requests.get(url)
        coin_data = response.json()

        # Check for rate limit response
        if isinstance(coin_data, dict) and coin_data.get('status', {}).get('error_code') == 429:
            logger.warning("Rate limit exceeded. Waiting 60 seconds...")
            time.sleep(60)
            yield from fetch_coins_recursively(page)
            return

        # Base case: empty response (end of pagination)
        if not coin_data:
            return

        # Process current page
        yield from coin_data
        logger.info(f"Fetched page {page} with {len(coin_data)} coins")

        # Be nice to the API
        time.sleep(1)

        # Recursive case: fetch next page
        yield from fetch_coins_recursively(page + 1)

    except requests.exceptions.RequestException as e:
        logger.error(f"Request failed on page {page}: {e}")
        raise

Then there is the get_full_coin_data_iteratively_for_page which is decorated by shared_task (for task autodiscovery). We supplied some parameters:

bind=True to access task instance via self
autoretry_for=(Exception,) to auto-retry on exceptions
retry_backoff=True for exponential backoff
max_retries=5 to limit retries to 5

For this task to be periodic, we must add it to the CELERY_BEAT_SCHEDULE in settings.py:

...
CELERY_BEAT_SCHEDULE: dict[str, dict[str, Any]] = {
    ...
    'get_full_coin_data_iteratively_for_page': {
        'task': 'core.tasks.get_full_coin_data_iteratively_for_page',
        'schedule': crontab(minute='*/3'),
    },
}

It schedules this task to run every 3 minutes ('*/3') using crontab.

These implementations were with performance in mind. However, there is still room for improvement.

Step 4: Bonus

There are APIs whose paginations are not page-based but use the next (default DRF pagination strategy). For these systems, the last bits of data have empty next. That's the breaking point:

def get_api_data(url: str) -> None:
    """Make recursive requests."""
    headers = {'Authorization': f'Token {settings.API_TOKEN}'}
    response = requests.get(url, headers=headers)
    data: dict[str, Any] = response.json()

    if 'next' in data and data.get('next') is not None:
        get_api_data(data.get('next'))

That's it! I hope you enjoyed it.

Outro

If you found this article valuable, consider sharing it with your network to help spread the knowledge!

Building an AI-powered Financial Behavior Analyzer with NodeJS, Python, SvelteKit, and TailwindCSS - Part 3: Transactions

John Owolabi Idogun — Thu, 06 Feb 2025 20:01:16 +0000

Introduction

Part 4 is already here: https://johnowolabiidogun.dev/blog/building-an-ai-powered-financial-behavior-analyzer-with-nodejs-python-sveltekit-and-tailwindcss-part-4-frontend-websocket-862be8/67a4fafe77db63877efad8cb

This series's previous article saw us build an OAuth authentication flow via GitHub. In this article, we'll continue adding APIs to the backend and specifically, implementing all the APIs related to transaction management while sticking with "our" modified MVC design pattern. Let's dive in.

Prerequite

We assume you've gone through the previous articles in this series where we built the app's utility server and the authentication flow. If not, kindly check them out. Also, this article requires that some modules be installed:

[x] busboy - module for parsing incoming HTML form data (handling file upload)
[x] csv-parse - CSV parser for Node.js and the web (latest version doesn't need additional @types/ installation)
[x] ws - a Node.js WebSocket library

The following command will install them as well as their types:

backend$ npm i busboy csv-parse ws # module installation
backend$ npm i -D @types/busboy @types/ws # types installation

Source code

Sirneij / finance-analyzer

An AI-powered financial behavior analyzer and advisor written in Python (aiohttp) and TypeScript (ExpressJS & SvelteKit with Svelte 5)

View on GitHub

Implementation

Note: Exclusion of interfaces/types

For brevity, we won't be including type declarations in this article and subsequent ones. You can always get them from the types directory.

Step 1: The parsers

We'll start by writing out our modular file parsers. There will be a shared BaseParser:

import type { IFileParser } from "$types/files.types.js";
import type { ITransaction, RawTransaction } from "$types/transaction.types.js";
import mongoose from "mongoose";

export abstract class BaseParser implements IFileParser {
  constructor(protected userId: mongoose.Types.ObjectId) {}

  abstract parse(buffer: Buffer): Promise<Partial<ITransaction>[]>;

  protected mapToTransaction(raw: RawTransaction): Partial<ITransaction> {
    return {
      userId: this.userId as mongoose.Types.ObjectId,
      date: new Date(raw.date),
      amount: Number(raw.amount),
      balance: raw.balance ? Number(raw.balance) : undefined,
      description: raw.description,
      type: raw.type.toLowerCase() === "income" ? "income" : "expense",
    };
  }
}

It is an abstract class that implements the IFileParser interface. It provides a shared structure for file parsers by:

Storing a user ID (needed for transaction filtering)
Declaring an abstract parse() method that concrete subclasses must implement to parse a file buffer into transactions.
Including a helper method, mapToTransaction(), which converts raw transaction data into a Partial<ITransaction> object, handling type conversions (e.g., converting a date string to a Date object, converting amount and balance to numbers, and ensuring the transaction type is in the correct string format).

All the other parsers would inherit this:

import { parse } from "csv-parse";
import { BaseParser } from "$utils/file.utils.js";
import type { ITransaction } from "$types/transaction.types.js";
import { baseConfig } from "$config/base.config.js";

export class CSVParser extends BaseParser {
  private readonly CHASE_COLUMNS = {
    details: /\b(details)\b/i,
    postingDate: /\b(posting[\s_-]?date)\b/i,
    description: /\b(description)\b/i,
    amount: /\b(amount)\b/i,
    type: /\b(type)\b/i,
    balance: /\b(balance)\b/i,
  };

  private cleanDescription(description: string): string {
    // Remove extra spaces
    let cleaned = description.replace(/\s+/g, " ").trim();

    // Remove location patterns
    cleaned = cleaned.replace(/\s+[A-Z]{2}\s*\d*$/, "");
    cleaned = cleaned.replace(/\s+\d{6}$/, "");

    // Remove date patterns
    cleaned = cleaned.replace(/\s+\d{2}\/\d{2}$/, "");

    // Remove special characters but keep basic punctuation
    cleaned = cleaned.replace(/[^\w\s-.,&()]/g, "");

    return cleaned || "Unknown Transaction";
  }

  private normalizeAmount(amount: string): number {
    // Convert string amount to number, preserving negative signs
    const value = parseFloat(amount.replace(/[^\d.-]/g, ""));
    return isNaN(value) ? 0 : value;
  }

  async parse(buffer: Buffer): Promise<Partial<ITransaction>[]> {
    return new Promise((resolve, reject) => {
      const transactions: Partial<ITransaction>[] = [];

      parse(buffer, {
        columns: true,
        trim: true,
        relax_column_count: true, // Allow inconsistent column counts
        skip_empty_lines: true,
        relax_quotes: false,
      })
        .on("data", (row) => {
          if (
            !row.Type ||
            !row.Amount ||
            !row["Posting Date"] ||
            !row.Description
          ) {
            throw new Error("Missing required fields");
          }
          const transactionType = row.Type.toLowerCase().includes("credit")
            ? "income"
            : "expense";
          const transaction = {
            userId: this.userId,
            date: new Date(row.Details ? row["Posting Date"] : Date.now()),
            amount: this.normalizeAmount(row.Amount),
            description: this.cleanDescription(row.Description),
            balance: row.Balance ? Number(row.Balance) : 0,
            type: transactionType as "income" | "expense",
          };

          if (!isNaN(transaction.amount)) {
            transactions.push(transaction);
          }
        })
        .on("error", (error) => {
          baseConfig.logger.error("Error parsing CSV", { error });

          return reject(error);
        })
        .on("end", () => {
          if (transactions.length === 0) {
            return reject(new Error("No valid transactions found"));
          }
          baseConfig.logger.info(`Parsed ${transactions.length} transactions`);
          resolve(transactions);
        });
    });
  }
}

This class extends the shared parser (BaseParser) to reuse common functionality while implementing CSV-specific logic. It uses the csv-parse library with options (like relaxed column count and skipping empty lines) to handle potentially inconsistent CSV data. While it implements helper functions (cleanDescription and normalizeAmount) to sanitize and convert raw CSV data, it throws an error immediately if required CSV fields are missing, ensuring data integrity. It also tries to dynamically determine the transaction type based on keywords in the CSV (credit implies income).

Next is the PDF parser:

import { baseConfig } from "$config/base.config.js";
import type { ITransaction } from "$types/transaction.types.js";
import { BaseParser } from "$utils/file.utils.js";

export class PDFParser extends BaseParser {
  private static readonly TRANSACTION_PATTERN =
    /(\d{2}\/\d{2})(?!.*Page)\s*([^-\d].*?)\s*(-\s*|\s+)?(\d{1,3}(?:,\d{3})*|\d+)\.(\d{2})\s*([\d,]+\.\d{2})?/;

  async parse(buffer: Buffer): Promise<Partial<ITransaction>[]> {
    const formData = new FormData();
    formData.append(
      "file",
      new Blob([buffer], { type: "application/pdf" }),
      "file.pdf"
    );

    const response = await fetch(
      `${baseConfig.utilityServiceUrl}/extract-text`,
      {
        method: "POST",
        body: formData,
      }
    );

    const data = (await response.json()) as { text: string };

    baseConfig.logger.info(`Parsed data: ${JSON.stringify(data)}`);

    const extractedText = data.text;
    const lines = extractedText.split("\n");

    const dateHeaderPattern = /([A-Za-z]+ \d{2}, (\d{4})) through/;
    let currentYear = new Date().getFullYear();

    const transactions: Partial<ITransaction>[] = [];

    for (let i = 0; i < lines.length; i++) {
      const line = lines[i].trim();

      const headerMatch = dateHeaderPattern.exec(line);
      if (headerMatch) {
        currentYear = parseInt(headerMatch[2]);
      }

      const match = PDFParser.TRANSACTION_PATTERN.exec(line);

      if (match) {
        const [, date, description, negative, whole, decimal, balance] = match;
        const amountStr = `${negative || ""}${whole}.${decimal}`;
        const parsedAmount = parseFloat(amountStr.replace(/[\s,]/g, ""));
        const parsedBalance = balance
          ? parseFloat(balance.replace(/,/g, ""))
          : 0.0;

        const fullDate = `${date}/${currentYear}`;

        transactions.push(
          this.mapToTransaction({
            userId: this.userId,
            date: fullDate,
            description: description.trim(),
            amount: parsedAmount,
            balance: parsedBalance,
            type: parsedAmount < 0 ? "expense" : "income",
          })
        );
      }
    }

    return transactions;
  }
}

As mentioned in the second article of this series, our AI service (utility) server will handle parsing PDF files since better tooling is available in the Python ecosystem for that, this parser uses an external API (our utility server API) to extract text from PDFs via a POST request. Then it updates the current year based on the date headers found in the text. Finally, it uses a regular expression to extract transaction details from each line and constructs transaction objects thereof.

To finish off the parser implementation, we need a nice way to automatically detect and use the right parser when a file is provided. For that, we have:

import mongoose from "mongoose";
import type { IFileParser, SupportedFileTypes } from "$types/files.types.js";
import { CSVParser } from "$utils/parsers/csv.parsers.js";
import { PDFParser } from "$utils/parsers/pdf.parsers.js";

export class ParserFactory {
  static getParser(
    mimeType: SupportedFileTypes,
    userId: mongoose.Types.ObjectId
  ): IFileParser {
    switch (mimeType) {
      case "text/csv":
        return new CSVParser(userId);
      case "application/pdf":
        return new PDFParser(userId);
      default:
        throw new Error(`Unsupported file type: ${mimeType}`);
    }
  }

  static isSupportedType(mimeType: string): mimeType is SupportedFileTypes {
    return [
      "text/csv",
      "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
      "application/pdf",
      "image/jpeg",
      "image/png",
    ].includes(mimeType);
  }
}

It implements a factory pattern that selects the correct parser based on the file's MIME type. The getParser() method returns a CSVParser or PDFParser instance (or errors for unsupported types), while isSupportedType() verifies if a MIME type is among the accepted ones.

Step 2: Transaction model

Before implementing other business logic, let's create a transaction schema and model so that MongoDB can help manage the data:

import type { ITransaction } from "$types/transaction.types.js";
import mongoose, { Schema } from "mongoose";

const transactionSchema = new Schema(
  {
    userId: {
      type: Schema.Types.ObjectId,
      ref: "User",
      required: true,
    },
    date: {
      type: Date,
      required: true,
    },
    amount: {
      type: Number,
      required: true,
    },
    balance: {
      type: Number,
      required: true,
    },
    description: {
      type: String,
      required: true,
    },
    type: {
      type: String,
      enum: ["income", "expense"],
      required: true,
    },
  },
  {
    timestamps: true,
  }
);

export const Transaction = mongoose.model<ITransaction>(
  "Transaction",
  transactionSchema
);

Just a simple schema with our defined columns.

Tip: Mongoose's automatic inclusion of date columns

When the timestamps is set to true, mongoose automatically adds createdAt and updatedAt to the schema. If you don't want that behavior, excluding it and giving it a false value is all you need.

Step 3: Business logic in `transaction.service.ts`

We'll proceed to writing our business logic in:

import { baseConfig } from "$config/base.config.js";
import { Transaction } from "$models/transaction.model.js";
import type {
  FileUploadResult,
  ITransaction,
  SpendingReport,
  FinancialSummary,
  PaginatedTransactions,
} from "$types/transaction.types.js";
import mongoose from "mongoose";
import { ParserFactory } from "$utils/parsers/factory.parsers.js";
import { WebSocket } from "ws";
import { sendError } from "$utils/error.utils.js";

export class TransactionService {
  static async processFile(
    buffer: Buffer,
    mimeType: string,
    userId: mongoose.Types.ObjectId
  ): Promise<FileUploadResult> {
    if (!ParserFactory.isSupportedType(mimeType)) {
      throw new Error("Unsupported file type");
    }

    const parser = ParserFactory.getParser(mimeType, userId);
    const data = await parser.parse(buffer);

    await Transaction.insertMany(data);

    return {
      mimeType,
      data: data,
    };
  }

  static async findTransactionsByUserId(
    userId: mongoose.Types.ObjectId,
    page: number = 1,
    limit: number = 9
  ): Promise<PaginatedTransactions> {
    try {
      // If limit is -1, fetch all transactions
      const shouldFetchAll = limit === -1;
      const skip = shouldFetchAll ? 0 : (page - 1) * limit;

      const [transactions, total] = await Promise.all([
        shouldFetchAll
          ? Transaction.find({ userId }).sort({ date: -1 }).lean().exec()
          : Transaction.find({ userId })
              .sort({ date: -1 })
              .skip(skip)
              .limit(limit)
              .lean()
              .exec(),
        Transaction.countDocuments({ userId }),
      ]);

      return {
        transactions,
        total,
        page: shouldFetchAll ? 1 : page,
        limit: shouldFetchAll ? total : limit,
      };
    } catch (error) {
      throw new Error("Failed to fetch transactions");
    }
  }

  static async summarizeTransactionsbyUserId(
    userId: mongoose.Types.ObjectId
  ): Promise<FinancialSummary> {
    try {
      const results = await this.findTransactionsByUserId(userId, 1, -1);
      const response = await fetch(
        `${baseConfig.utilityServiceUrl}/summarize`,
        {
          method: "POST",
          headers: {
            "Content-Type": "application/json",
          },
          body: JSON.stringify(results.transactions),
        }
      );

      if (!response.ok) {
        throw new Error("Failed to summarize transactions");
      }

      return response.json() as Promise<FinancialSummary>;
    } catch (error) {
      throw new Error("Failed to fetch income/expenses/savings");
    }
  }

  static async analyzeTransactionsByUserId(
    userId: mongoose.Types.ObjectId
  ): Promise<SpendingReport> {
    try {
      const transactions = await this.findTransactionsByUserId(userId, 1, -1);
      const response = await fetch(`${baseConfig.utilityServiceUrl}/analyze`, {
        method: "POST",
        headers: {
          "Content-Type": "application/json",
        },
        body: JSON.stringify(transactions),
      });

      if (!response.ok) {
        throw new Error("Failed to analyze transactions");
      }

      return response.json() as Promise<SpendingReport>;
    } catch (error) {
      throw new Error("Failed to analyze transactions");
    }
  }

  static async createTransactionsByUserId(
    userId: mongoose.Types.ObjectId,
    transactions: ITransaction[]
  ): Promise<void> {
    try {
      await Transaction.insertMany(transactions.map((t) => ({ ...t, userId })));
    } catch (error) {
      throw new Error("Failed to create transactions");
    }
  }

  static async deleteTransactionsByUserId(
    userId: mongoose.Types.ObjectId,
    transactionIds: mongoose.Types.ObjectId[]
  ): Promise<void> {
    try {
      await Transaction.deleteMany({ userId, _id: { $in: transactionIds } });
    } catch (error) {
      throw new Error("Failed to delete transactions");
    }
  }

  static async connectToUtilityServer(
    action: string,
    transactions: ITransaction[],
    frontendWs: WebSocket
  ) {
    const wsUrl = baseConfig.utilityServiceUrl.replace(/^http/, "ws");
    const ws = new WebSocket(`${wsUrl}/ws`);

    ws.on("open", () => {
      baseConfig.logger.info(
        `Connected to utility server for '${action}' at ${wsUrl}`
      );
      ws.send(JSON.stringify({ action, transactions }));
    });

    ws.on("message", (message: string) => {
      const data = JSON.parse(message);
      frontendWs.send(JSON.stringify(data));
    });

    ws.on("close", () => {
      frontendWs.send(
        JSON.stringify({
          action: "progress",
          message: `Connection to utility server closed for ${action}.`,
          type: action,
        })
      );
    });

    ws.on("error", (err) => {
      sendError(
        frontendWs,
        `Utility server WebSocket error: ${err.message}`,
        action
      );
    });
  }
}

This service class orchestrates all transaction-related operations by:

Parsing files with the appropriate parser (via ParserFactory) and inserting the parsed transactions.
Fetching (with pagination) transactions, summarizing financial data, and analyzing spending via our utility service endpoints.
Creating and deleting transactions.
Establishing a WebSocket connection to forward real-time updates from our utility server to the front end.

Tip: Better query fetching with pagination

For fetching transaction data with pagination, we are currently using two queries: find and countDocuments. We can optimize this by using a single query aggregation pipeline.

Next, we write the controllers.

Step 4: Transactions controllers

Here we provide the controllers for the transaction:

import { baseConfig } from "$config/base.config.js";
import { TransactionService } from "$services/transaction.service.js";
import busboy from "busboy";
import type { Request, Response } from "express";
import mongoose from "mongoose";

export class TransactionController {
  async handleFileUpload(req: Request, res: Response): Promise<void> {
    try {
      const userId = req.user?._id as mongoose.Types.ObjectId;

      await new Promise<void>((resolve, reject) => {
        const bb = busboy({ headers: req.headers });
        let isFileProcessed = false;

        bb.on("file", (name, file, info) => {
          // TODO: Handle multiple files and excel files later
          // But for now, we only handle one file and csv or pdf file
          // so reject if there are multiple files or non-csv/pdf files
          if (isFileProcessed) {
            reject(new Error("Only one file is allowed"));
            return;
          }
          if (
            info.mimeType !== "text/csv" &&
            info.mimeType !== "application/pdf"
          ) {
            reject(new Error("Only CSV and PDF files are allowed"));
            return;
          }

          const chunks: Buffer[] = [];

          file.on("data", (chunk) => chunks.push(chunk));

          file.on("end", async () => {
            try {
              const buffer = Buffer.concat(chunks);
              const result = await TransactionService.processFile(
                buffer,
                info.mimeType,
                userId
              );
              baseConfig.logger.info(`Result: ${JSON.stringify(result)}`);
              if (!isFileProcessed) {
                isFileProcessed = true;
                res.json({ success: true, ...result });
                resolve();
              }
            } catch (err) {
              baseConfig.logger.error(`Catch: ${err}`);
              reject(err);
            }
          });
        });

        bb.on("error", (err) => {
          baseConfig.logger.error(`OnError: ${err}`);
          reject(err);
        });

        req.pipe(bb);
      });
    } catch (error) {
      res.status(400).json({
        success: false,
        message: error instanceof Error ? error.message : "Upload failed",
      });
    }
  }

  async handleGetTransactions(req: Request, res: Response): Promise<void> {
    try {
      const userId = req.user?._id as mongoose.Types.ObjectId;

      let page = Number(req.query.page);
      let limit = Number(req.query.limit);

      if (isNaN(page)) {
        page = 1;
      }

      if (isNaN(limit)) {
        limit = 10;
      }

      const result = await TransactionService.findTransactionsByUserId(
        userId,
        page,
        limit
      );

      res.json({
        transactions: result.transactions,
        metadata: {
          total: result.total,
          page: result.page,
          limit: result.limit,
          totalPages: Math.ceil(result.total / result.limit),
        },
        success: true,
      });
    } catch (error) {
      res.status(400).json({
        success: false,
        message:
          error instanceof Error
            ? error.message
            : "Failed to fetch transactions",
      });
    }
  }

  async handleGetIncomeExpensesSavings(
    req: Request,
    res: Response
  ): Promise<void> {
    try {
      const userId = req.user?._id as mongoose.Types.ObjectId;

      const summary = await TransactionService.summarizeTransactionsbyUserId(
        userId
      );

      res.status(200).json({ summary, success: true });
    } catch (error) {
      res.status(400).json({
        success: false,
        message:
          error instanceof Error
            ? error.message
            : "Failed to fetch income/expenses/savings",
      });
    }
  }

  async handleAnalyzeTransactions(req: Request, res: Response): Promise<void> {
    try {
      const userId = req.user?._id as mongoose.Types.ObjectId;

      const data = await TransactionService.analyzeTransactionsByUserId(userId);

      res.status(200).json({ data, success: true });
    } catch (error) {
      res.status(400).json({
        success: false,
        message:
          error instanceof Error
            ? error.message
            : "Failed to analyze transactions",
      });
    }
  }

  async handleCreateTransactions(req: Request, res: Response): Promise<void> {
    try {
      const userId = req.user?._id as mongoose.Types.ObjectId;

      const transactions = req.body;

      await TransactionService.createTransactionsByUserId(userId, transactions);

      res.status(200).json({ success: true });
    } catch (error) {
      res.status(400).json({
        success: false,
        message:
          error instanceof Error
            ? error.message
            : "Failed to create transactions",
      });
    }
  }

  async handleDeleteTransactions(req: Request, res: Response): Promise<void> {
    try {
      const userId = req.user?._id as mongoose.Types.ObjectId;

      const transactionIds = req.body;

      await TransactionService.deleteTransactionsByUserId(
        userId,
        transactionIds
      );

      res.json({ success: true });
    } catch (error) {
      res.status(400).json({
        success: false,
        message:
          error instanceof Error
            ? error.message
            : "Failed to delete transactions",
      });
    }
  }
}

This controller class handles HTTP requests for transaction-related operations. It processes file uploads (using busboy) by delegating to TransactionService, and exposes endpoints for fetching (with pagination), summarizing, analyzing, creating, and deleting transactions—all with appropriate error handling.

Step 5: Routes and configuration

The last step is coupling this together. Let's first create the routes:

import { TransactionController } from "$controllers/transaction.controller.js";
import { isAuthenticated } from "$middlewares/auth.middleware.js";
import { Router } from "express";

const transactionRoutes = Router();
const transactionController = new TransactionController();

transactionRoutes.get(
  "/",
  isAuthenticated,
  transactionController.handleGetTransactions
);

transactionRoutes.post(
  "/",
  isAuthenticated,
  transactionController.handleCreateTransactions
);

transactionRoutes.delete(
  "/",
  isAuthenticated,
  transactionController.handleDeleteTransactions
);

transactionRoutes.post(
  "/upload",
  isAuthenticated,
  transactionController.handleFileUpload
);

transactionRoutes.get(
  "/summary",
  isAuthenticated,
  transactionController.handleGetIncomeExpensesSavings
);

transactionRoutes.get(
  "/analyze",
  isAuthenticated,
  transactionController.handleAnalyzeTransactions
);

export default transactionRoutes;

And lastly, register it with with app:

...
import transactionRoutes from "$routes/transaction.routes.js";
...
// Authentication routes
app.use("/api/v1/auth", authRoutes);
app.use(handleAuthError);

// Transaction routes
app.use("/api/v1/transactions", transactionRoutes);
...
startServer();

We will stop here in this article. In the next one, we will complete this transaction service by writing the WebSocket handler. Then, we will incept the front end with SvelteKit by implementing the authentication flow, and user dashboard with TailwindCSS v4.

Outro

Enjoyed this article? I'm a Software Engineer and Technical Writer actively seeking new opportunities, particularly in areas related to web security, finance, healthcare, and education. If you think my expertise aligns with your team's needs, let's chat! You can find me on LinkedIn and X. I am also an email away.

If you found this article valuable, consider sharing it with your network to help spread the knowledge!

Building an AI-powered Financial Behavior Analyzer with NodeJS, Python, SvelteKit, and TailwindCSS - Part 2: GitHub OAuth

John Owolabi Idogun — Wed, 05 Feb 2025 18:22:53 +0000

Introduction

Part 3 is already out here: https://johnowolabiidogun.dev/blog/building-an-ai-powered-financial-behavior-analyzer-with-nodejs-python-sveltekit-and-tailwindcss-part-3-transactions-0da982/67a3a71d72d0dd09224068e8

With the AI service completed, we now focus on setting up the base backend service. As outlined in the system's architecture, our main backend will be powered by Node.js with Express.js, using MongoDB as the database.

While we could have chosen other combinations—such as Node.js with PostgreSQL, Bun with PostgreSQL, or Django/Flask/FastAPI/aiohttp with SQL/NoSQL/NewSQL—I opted for this stack as a refresher.

We'll begin developing the backend service here and continue refining it in subsequent articles. Let's dive in! 🚀

Prerequisite

We assume you have already set up a TypeScript-based Express project. If not, follow these simple steps:

[x] Create a new folder, say backend, and change the directory into it
[x] Run npm init -y to initialize a node project:

  backend$ npm init -y

This will create a normal package.json file with very basic entries.

[x] Install TypeScript and create a tsconfig.json file sh backend$ npm install --save-dev typescript && npx tsc --init

At this point, you have a minimal Node.js app with TypeScript support. However, for our project, we need a more structured setup. Modify your package.json as follows:

{
  "name": "ai-powered-financial-behavior-analyzer",
   "version": "1.0.0",
  "description": "An API for analyzing financial behavior using AI",
  "main": "dist/app.js",
  "scripts": {
    "dev": "tsx watch src/app.ts",
    "build": "tsc && tsc-alias",
    "start": "node dist/src/app.js",
    "test": "jest --config jest.config.js",
    "test:coverage": "jest --config jest.config.js --coverage --silent=false"
  },
  "keywords": [],
  "author": "John Owolabi Idogun",
  "license": "ISC",
  "devDependencies": {
    "typescript": "^5.7.3"
  },
  "type": "module",
  "engines": {
    "node": "22.x"
  },
  "jest": {
    "extensionsToTreatAsEsm": [
      ".ts"
   ]
  }
}

We want to use ESM instead of CommonJS hence the "type": "module". We also want to use the latest node.js LTS (v22 at the time of writing). We also want some structure where all source files are in the src directory and tests in the tests directory with the entry point, during development, as src/app.ts (in production, it'll be dist/app.js). Next, make tsconfig.json look like this:

{
  "compilerOptions": {
    "target": "ES2023",
    "module": "NodeNext",
    "moduleResolution": "nodenext",
    "strict": true,
    "esModuleInterop": true,
    "skipLibCheck": true,
    "outDir": "./dist",
    "rootDir": ".",
    "baseUrl": ".",
    "allowJs": true,
    "resolveJsonModule": true,
    "allowImportingTsExtensions": false,
    "typeRoots": ["./node_modules/@types", "./src/types"],
    "paths": {
      "$config/*": ["src/config/*"],
      "$controllers/*": ["src/controllers/*"],
      "$models/*": ["src/models/*"],
      "$routes/*": ["src/routes/*"],
      "$services/*": ["src/services/*"],
      "$types/*": ["src/types/*"],
      "$middlewares/*": ["src/middlewares/*"],
      "$utils/*": ["src/utils/*"],
      "$websockets/*": ["src/websockets/*"]
    },
    "experimentalDecorators": true,
    "emitDecoratorMetadata": true
  },
  "include": ["src/**/*", "test/**/*.ts", "jest.setup.js"],
  "exclude": ["node_modules", "dist"]
}

You can get a better explanation for each of the entries here. The idea is we want to use the latest entries while being considerate. We also created aliases (that's what paths does) so that instead of doing ../../../src/config/base.js, we would just do: $config/base.js. Nifty stuff!

Source code

Sirneij / finance-analyzer

An AI-powered financial behavior analyzer and advisor written in Python (aiohttp) and TypeScript (ExpressJS & SvelteKit with Svelte 5)

View on GitHub

Implementation

Now it's time to get our hands dirty. We will be implementing the OAuth-based authentication system here.

Step 1: Install dependencies and setup configurations

We will use passports.js and passport-github2 for implementing the authentication strategy. Let's install the libraries (and types to keep TypeScript happy):

# Packages
backend$ npm i express@^5.0.1 passport passport-github2 mongoose cors redis connect-redis express-session dotenv winston

# Types
backend$ npm i -D @types/express  @types/passport  @types/passport-github2  @types/mongoose  @types/cors  @types/redis  @types/connect-redis  @types/express-session  @types/dotenv @types/winston

Note: Create a GitHub OAuth App

As with most OAuth services, you need to create a new GitHub OAuth app to use its authentication strategy. After creation, you will be provided with GITHUB_CLIENT_ID and GITHUB_CLIENT_SECRET which are required by passport-github2. Ensure you fill in your app details correctly.

Having installed the packages, let's start with some configurations. Create src/types/misc.types.ts and populate it with:

import { AuthConfig } from "$types/auth.types.js";
import { DbConfig } from "$types/db.types.js";
import winston from "winston";

export enum Providers {
  GOOGLE = 1,
  GITHUB = 2,
}

export interface BaseConfig {
  auth: AuthConfig;
  db: DbConfig;
  frontendUrl: string;
  utilityServiceUrl: string;
  redisUrl: string;
  logger: winston.Logger;
}

Though we are only supporting GitHub OAuth for now, we defined an enum that includes Google as well. The BaseConfig interface will be used for all the app's configurations including auth and db which have standalone types:

export interface OAuthCredentials {
  clientID: string;
  clientSecret: string;
  callbackURL: string;
}

export interface AuthConfig {
  google: OAuthCredentials;
  github: OAuthCredentials;
  session: {
    secret: string;
  };
}

```ts :src/types/db.types.ts:
export interface DbConfig {
uri: string;
dbName: string;
}




As previously stated, we need each OAuth service's `cliend_id` and `client_secret` to authenticate. Another important credential is the `callback_url` (also called `redirect URI` in some OAuth implementations) which is the URL where the OAuth service redirects users after authentication. You will supply this at the point of registering or creating a new OAuth app in GitHub (and other providers as well).

Next, let's find a way to populate these credentials. We will use the `dotenv` package to retrieve them from a `.env` file or environment variables:



```ts
import { BaseConfig } from "$types/misc.types.js";
import { authConfig } from "$config/internal/auth.config.js";
import { dbConfig } from "$config/internal/db.config.js";
import { logger } from "$config/internal/logger.config.js";

export const baseConfig: BaseConfig = {
  get frontendUrl() {
    return process.env.FRONTEND_URL || "http://localhost:3000";
  },
  get utilityServiceUrl() {
    return process.env.UTILITY_SERVICE_URL || "http://localhost:5173";
  },
  get redisUrl() {
    return process.env.REDIS_URL || "redis://localhost:6379";
  },
  auth: authConfig,
  db: dbConfig,
  logger:
};

We use getter methods in our configuration for three key benefits:

Dynamic Values: Getters retrieve values on-demand, ensuring we always get the latest values
Environment Variables: Particularly important for process.env values that may change during runtime
Lazy Evaluation: Values are only computed when accessed, improving performance

I personally encountered a bug in production where my authentication process was failing because stale values were being read by process.env for frontendUrl.

Here are the contents of src/config/internal/auth.config.ts:

import { AuthConfig } from "$types/auth.types.js";
import { config } from "dotenv";

config();

export const authConfig: AuthConfig = {
  google: {
    clientID: process.env.GOOGLE_CLIENT_ID!,
    clientSecret: process.env.GOOGLE_CLIENT_SECRET!,
    callbackURL: `${process.env.APP_URL}/api/v1/auth/google/callback`,
  },
  github: {
    clientID: process.env.GITHUB_CLIENT_ID!,
    clientSecret: process.env.GITHUB_CLIENT_SECRET!,
    callbackURL: `${process.env.APP_URL}/api/v1/auth/github/callback`,
  },
  session: {
    secret: process.env.SESSION_SECRET || "your-secret-key",
  },
};

import { DbConfig } from "$types/db.types.js";

import { config } from "dotenv";

config();

export const dbConfig: DbConfig = {
  uri: process.env.DB_URI || "mongodb://localhost:27017",
  dbName: process.env.DB_NAME || "test",
};

import winston from "winston";

// Colors for different log levels
const colors = {
  error: "red",
  warn: "yellow",
  info: "green",
  http: "magenta",
  debug: "blue",
};

winston.addColors(colors);

// Development format - pretty console output
const developmentFormat = winston.format.combine(
  winston.format.colorize(),
  winston.format.timestamp({ format: "YYYY-MM-DD HH:mm:ss" }),
  winston.format.printf(
    (info) => `${info.timestamp} ${info.level}: ${info.message}`
  )
);

// Production format - JSON for better parsing
const productionFormat = winston.format.combine(
  winston.format.timestamp(),
  winston.format.json()
);

export const logger = winston.createLogger({
  level: process.env.NODE_ENV === "development" ? "debug" : "info",
  format:
    process.env.NODE_ENV === "development"
      ? developmentFormat
      : productionFormat,
  transports: [new winston.transports.Console()],
});

Tip: Use MongoDB Atlas

You probably have MongoDB installed on your machine and you can just use it for development. In case you need to test in production, check up the free version of the MongoDB Atlas.

Step 2: Connect to MongoDB database and redis

With the configurations underway, let's create services that will connect our application to MongoDB as well as redis (for session storage).

<span><strong>Note:</strong> Why Redis for Session Storage?</span>
<p>
    While <code>express-session</code> offers in-memory storage, Redis is preferred for production because:
</p>
<ul>
    <li>

Persistence: Sessions survive server restarts

Scalability: Handles high traffic and multiple server instances

Performance: Fast read/write operations with minimal latency

Memory Management: Automatic memory optimization and key expiration

Using in-memory storage in production can lead to:

Lost sessions after server restarts
Memory leaks as sessions accumulate
Scaling issues with multiple server instances

Let's create a database service for them:

import mongoose from "mongoose";
import { baseConfig } from "$config/base.config.js";
import { RedisStore } from "connect-redis";
import { createClient } from "redis";

const MAX_RETRIES = 3;
const RETRY_INTERVAL = 5000;

eexport async function connectToCluster(retryCount = 0) {
  try {
    const options = {
      dbName: baseConfig.db.dbName,
      serverSelectionTimeoutMS: 15000,
      socketTimeoutMS: 45000,
      maxPoolSize: 50,
      minPoolSize: 10,
      retryWrites: true,
      retryReads: true,
    };
    await mongoose.connect(baseConfig.db.uri, options);

    mongoose.connection.on("error", (err) => {
      baseConfig.logger.error("❌ MongoDB connection error:", err);
    });

    mongoose.connection.once("open", () => {
      baseConfig.logger.info("✅ MongoDB connection successful");
    });
    // Handle graceful shutdown
    process.on("SIGINT", async () => {
      try {
        await mongoose.connection.close();
        baseConfig.logger.info("MongoDB connection closed");
        process.exit(0);
      } catch (err) {
        baseConfig.logger.error("Error closing MongoDB connection:", err);
        process.exit(1);
      }
    });

    return mongoose.connection;
  } catch (error) {
    baseConfig.logger.error("❌ MongoDB connection error:", error);

    if (retryCount < MAX_RETRIES) {
      baseConfig.logger.info(
        `Retrying connection to MongoDB cluster in ${
          RETRY_INTERVAL / 1000
        } seconds...`
      );
      await new Promise((resolve) => setTimeout(resolve, RETRY_INTERVAL));
      return connectToCluster(retryCount + 1);
    }

    throw error;
  }
}

export const connectToRedis = (): RedisStore => {
  const redisClient = createClient({
    url: baseConfig.redisUrl,
  });

  redisClient.connect().catch((error) => {
    baseConfig.logger.error("❌ Redis connection error:", error);
  });

  redisClient.on("connect", () => {
    baseConfig.logger.info("✅ Redis connection successful");
  });

  return new RedisStore({ client: redisClient, prefix: "session:" });
};

For the MongoDB connection, we implemented a retry logic in case some connection attempts fail. Aside from that, it's a basic way to connect to a MongoDB instance. We did something equivalent to redis. Now, we can proceed to hook all these up in src/app.ts.

Step 3: Setting up an express server in `src/app.ts`

Let's populate our src/app.ts with the following:

import express, { Application } from "express";
import cors from "cors";
import session from "express-session";
import passport from "passport";
import { Strategy as GitHubStrategy } from "passport-github2";
import { connectToCluster, connectToRedis } from "$services/db.service.js";
import { baseConfig } from "$config/base.config.js";
import { AuthService } from "$services/auth.service.js";
import authRoutes from "$routes/auth.routes.js";
import { Providers } from "$types/misc.types.js";
import { GitHubProfile } from "$types/auth.types.js";
import type { User } from "$types/passports.d.js";
import { ProviderMismatchError } from "$types/error.types.js";
import { createServer, Server as HttpServer } from "http";

const app: Application = express();

// 1. Trust proxy setting
app.set("trust proxy", 1);

// 2. Basic middleware
app.use(express.json());

// 3. CORS configuration
app.use(
  cors({
    origin: baseConfig.frontendUrl,
    credentials: true,
    methods: [
      "GET",
      "POST",
      "PUT",
      "PATCH",
      "DELETE",
      "OPTIONS",
      "HEAD",
      "TRACE",
      "CONNECT",
    ],
    allowedHeaders: ["Content-Type", "Authorization"],
    exposedHeaders: ["set-cookie"],
  })
);

// 4. Session configuration
app.use(
  session({
    store: baseConfig.redisUrl ? connectToRedis() : new session.MemoryStore(),
    secret: baseConfig.auth.session.secret,
    resave: false,
    saveUninitialized: false,
    proxy: true,
    cookie: {
      secure: process.env.NODE_ENV === "production",
      sameSite: process.env.NODE_ENV === "production" ? "none" : "lax",
      httpOnly: true,
      maxAge: 24 * 60 * 60 * 1000,
      domain:
        process.env.NODE_ENV === "production"
          ? baseConfig.cookieDomain
          : undefined,
    },
  })
);

// 5. Authentication middleware
app.use(passport.initialize());
app.use(passport.session());

passport.serializeUser<User>((user, done) => {
  done(null, user);
});

passport.deserializeUser((user: User, done) => {
  done(null, user);
});

passport.use(
  new GitHubStrategy(
    baseConfig.auth.github,
    async (
      accessToken: any,
      refreshToken: any,
      profile: GitHubProfile,
      done: (error: any, user?: any, options?: { message: string }) => void
    ) => {
      try {
        const user = await AuthService.findOrCreateUser({
          id: profile.id,
          email: profile.emails?.[0].value,
          provider: profile.provider,
          providerId: Providers.GITHUB,
          avatar: profile.photos?.[0].value,
          name: profile.displayName,
        });
        return done(null, user);
      } catch (error) {
        if (error instanceof ProviderMismatchError) {
          return done(null, false, { message: error.message });
        }
        return done(error);
      }
    }
  )
);

// Authentication routes
app.use("/api/v1/auth", authRoutes);

// Health check
app.get("/api/v1/health", (req, res) => {
  baseConfig.logger.info("Health check endpoint called");
  res.status(200).json({ message: "Server is running" });
});

const startServer = async () => {
  try {
    const server: HttpServer = createServer(app);

    const db = await connectToCluster();

    if (!db.readyState) {
      throw new Error("MongoDB connection not ready");
    }

    // 7. Start server
    const PORT = process.env.PORT || 3000;
    server.listen(PORT, () => {
      baseConfig.logger.info(`Server listening on port ${PORT}`);
    });
  } catch (error) {
    baseConfig.logger.error("Error starting server:", error);
    process.exit(1);
  }
};
startServer();

It is a simple setup that also considers production environments. We started by creating an express application instance which is needed to attach. Next, we enabled the "trust proxy" which allows some features for applications behind a proxy. Then we informed Express to parse incoming requests with JSON payloads by "using" the express.json() middleware. This is a way of attaching middleware to express. We also used the CORS and session middlewares to appropriately configure our application's CORS for inter-origin resource sharing and sessions. Specifically, we are giving the backend a "go-ahead" to share resources with our front end. We also made sure our sessions were secured by providing them with an option to use our generated secret keys. In development, you can use openssl to generate a 32-bit secret key:

$ openssl rand -base64 32

In production, you can opt for Cryptographically generated bytes.

After that, we used passport's authentication middleware and extended its feature to easily serialize and deserialize our app's User object. In the spirit of authentication, we defined our GitHub OAuth strategy next and it follows the normal anatomy of OAuth strategies supported by passport and specifically, passport-github2. Because we defined our credentials perfectly, we just passed it in, else we would have done something like:

passport.use(new GitHubStrategy({
    clientID: process.env['HITHUB_CLIENT_ID'],
    clientSecret: process.env['GITHUB_CLIENT_SECRET'],
    callbackURL: 'https://www.example.com/oauth2/redirect/google'
  },
  ...

In the callback function, we have access to the profile data returned by GitHub which was then passed into the AuthService to create the user in the database:

import { User } from "$models/user.model.js";
import { AuthUser, UserProfile } from "$types/auth.types.js";
import { ProviderMismatchError } from "$types/error.types.js";

export class AuthService {
  static async findOrCreateUser(profile: UserProfile) {
    try {
      // First try to find user by email only
      let user = await User.findOne({ email: profile.email }).exec();

      if (user) {
        // User exists, check provider
        if (user.provider !== profile.provider) {
          throw new ProviderMismatchError(user.provider);
        }

        // Check if any details need updating
        const updates: Partial<AuthUser> = {};
        if (user.name !== profile.name) updates.name = profile.name;
        if (user.providerId !== profile.providerId)
          updates.providerId = profile.providerId;
        if (user.avatar !== profile.avatar) updates.avatar = profile.avatar;
        // If updates needed, apply them
        if (Object.keys(updates).length > 0) {
          user = await User.findByIdAndUpdate(
            user._id,
            { $set: updates },
            { new: true }
          ).exec();
        }
      } else {
        // Create new user if none exists
        user = await User.create({
          email: profile.email,
          name: profile.name,
          provider: profile.provider,
          providerId: profile.providerId,
          avatar: profile.avatar,
        });
      }

      return user;
    } catch (error) {
      console.error("Error in findOrCreateUser:", error);
      throw error;
    }
  }
}

We defined a user model (with its schema) already:

import mongoose, { Schema } from "mongoose";
import { AuthUser } from "$types/auth.types.js";

const userSchema = new Schema<AuthUser>(
  {
    email: { type: String, required: true, unique: true },
    name: { type: String },
    provider: { type: String, required: true },
    providerId: { type: Number, required: true },
    avatar: { type: String, default: null },
  },
  {
    timestamps: true,
  }
);

export const User = mongoose.model<AuthUser>("User", userSchema);

The types used so far for the user can be found in src/types/auth.types.ts:

import mongoose from "mongoose";

export interface UserProfile {
  id: string;
  email: string;
  name?: string;
  provider: "google" | "github";
  providerId: number;
  avatar: string | null;
}

export interface AuthUser {
  _id?: mongoose.Types.ObjectId;
  email: string;
  name?: string;
  provider: string;
  providerId: number;
  avatar: string | null;
  isJohnOwolabiIdogun: boolean;
  createdAt: Date;
  updatedAt: Date;
}
...

To make TypeScript happy with our custom user type, we needed to modify its user type in src/types/passport.d.ts:

import { AuthUser } from "$types/auth.types.js";

declare global {
  namespace Express {
    interface User extends AuthUser {}
  }
}

// Re-export the User type
export type User = Express.User;

The rest of the src/app.ts are pretty basic. Before we wrap up with this article, let's see what the authentication routes are.

Step 4: Authentication routes

In src/app.ts, we used:

...
import { AuthService } from "$services/auth.service.js";
import authRoutes from "$routes/auth.routes.js";
...

// Authentication routes
app.use("/api/v1/auth", authRoutes);
...

These routes are in src/routes/auth.routes.ts:

import { Router } from "express";
import passport from "passport";
import { AuthController } from "$controllers/auth.controller.js";
import { isAuthenticated } from "$middlewares/auth.middleware.js";

const authRouters = Router();
const authController = new AuthController();

authRouters.get("/github", (req, res, next) => {
  const state = req.query.next
    ? Buffer.from(req.query.next as string).toString("base64")
    : "";
  passport.authenticate("github", {
    scope: ["user:email"],
    state,
  })(req, res, next);
});

authRouters.get(
  "/github/callback",
  passport.authenticate("github", { failureRedirect: "/api/v1/auth/failure" }),
  (req, res, next) => {
    next();
  },
  authController.handleLoginSuccess
);

authRouters.get("/session", isAuthenticated, (req, res) => {
  res.json({ user: req.user });
});

authRouters.get("/failure", authController.handleLoginFailure);
authRouters.get("/logout", authController.handleLogout);

export default authRouters;

The first one is where the authentication flow starts. It lets you login into your GitHub account and if successful redirects you to the callback_url the developer supplied during OAuth app creation, for us, it's the second route.

Tip: Supplying redirect route in the frontend

Let's say a user wants to access /private/route in your app's front end but such a user wasn't authenticated. Then your frontend app redirects the user to login with GitHub (and provides a next=/private/route in the URL). What the user expects is after a successful login, they want to be sent back to where they were headed initially /private/route. That was the logic implemented in the /github route above. It simply "remembers" the user's previous state.

These routes are very basic. We won't talk much about them. However, they used some "controllers" which we haven't seen yet:

import { baseConfig } from "$config/base.config.js";
import { Request, Response } from "express";

export class AuthController {
  async handleLoginSuccess(req: Request, res: Response) {
    if (req.user) {
      if (req.xhr || req.headers.accept?.includes("application/json")) {
        res.status(200).json({
          success: true,
          message: "Login successful",
          user: req.user,
        });
      } else {
        const state = req.query.state as string | undefined;
        let redirectPath = "/";

        if (state) {
          try {
            // Validate if the state is Base64
            const base64Regex = /^[A-Za-z0-9+/=]+$/;
            if (base64Regex.test(state)) {
              redirectPath = Buffer.from(state, "base64").toString();
            } else {
              throw new Error("Invalid Base64 input");
            }
          } catch (error) {
            baseConfig.logger.error("Failed to decode state parameter:", error);
          }
        }
        baseConfig.logger.info(
          `Redirecting to ${baseConfig.frontendUrl}${redirectPath}`
        );
        res.redirect(`${baseConfig.frontendUrl}${redirectPath}`);
      }
    }
  }

  async handleLoginFailure(req: Request, res: Response) {
    baseConfig.logger.info(
      `Redirecting to ${baseConfig.frontendUrl}/finanalyzer/auth/login?error=true`
    );
    res.redirect(`${baseConfig.frontendUrl}/finanalyzer/auth/login?error=true`);
  }

  async handleLogout(req: Request, res: Response) {
    req.logout(() => {
      baseConfig.logger.info(
        `Redirecting to ${baseConfig.frontendUrl}/finanalyzer/auth/login`
      );
      res.redirect(`${baseConfig.frontendUrl}/finanalyzer/auth/login`);
    });
  }
}

We redirected responses back to our frontend app.

With that, I will say see you in the next release! Check out the GitHub repository for the other missing pieces.

Outro

If you found this article valuable, consider sharing it with your network to help spread the knowledge!

Building an AI-powered Financial Behavior Analyzer with NodeJS, Python, SvelteKit, and TailwindCSS - Part 1: The AI Service

John Owolabi Idogun — Mon, 03 Feb 2025 07:03:40 +0000

Introduction

Part 2: GitHub OAuth is out on https://johnowolabiidogun.dev/blog/building-an-ai-powered-financial-behavior-analyzer-with-nodejs-python-sveltekit-and-tailwindcss-part-2-github-oauth-5195c7/67a06839edc9a2f58232fbf8

To kick off this intriguing and challenging journey, we'll start by building the AI service, the final component of our system’s architecture. We'll exploit the power of asynchronous programming in Python via the asyncio and one of its most lightweight yet powerful asynchronous HTTP Client/Servers, aiohttp. It is a "double-edged" sword that can be your http/websocket client and at the same time, http/websocket server. Unlike requests, which is synchronous, aiohttp supports both HTTP and WebSocket communication, making it a worthy alternative to FastAPI, Flask, or even Django for async-based applications.

Let's put aiohttp to work in this article! 🚀

Prerequisite

Before diving in, I assume you've already:

[x] Created a project (e.g., utility)
[x] Set up a virtual environment
[x] Installed the required dependencies

If you haven't, copy the following into requirements.txt:

aiohttp==3.11
pandas==2.2
pdf2image==1.17
pytesseract==0.3
transformers==4.48
python-dotenv==1.0
scikit-learn==1.6
torch @ https://download.pytorch.org/whl/cpu/torch-2.5.1%2Bcpu-cp312-cp312-linux_x86_64.whl
torchvision @ https://download.pytorch.org/whl/cpu/torchvision-0.20.1%2Bcpu-cp312-cp312-linux_x86_64.whl

and, while in your project's folder's virtual environment, run:

(virtualenv)$ pip install -r requirements.txt

Notice the versions of torch and torchvision used here. I used CPU-only versions of torch and torchvision since I deployed this on a CPU-based server. If you have a GPU-enabled server or development environment, you can install the standard versions without the URL specifier (@ https://download.pytorch.org/whl/...) which, by the way, is a nifty way to include libraries' URL in requirements.txt.

Note: You may prefer to develop using the latest tourch (v2.6).

Just as I write this, The PyTorch Foundation announced the release of PyTorch® 2.6 which debuts it for Python 3.13 with some performance improvements. Consider using it instead.

Source code

Sirneij / finance-analyzer

An AI-powered financial behavior analyzer and advisor written in Python (aiohttp) and TypeScript (ExpressJS & SvelteKit with Svelte 5)

View on GitHub

Implementation

Let's now go into the meat of this article.

Step 1: Spawning an `aiohttp` Server

Create app.py and populate it with the following:

import os
from asyncio import Lock

from aiohttp import WSMsgType, web
from aiohttp.web import Request, Response, WebSocketResponse

from utils.analyzer import analyze_transactions
from utils.extract_text import extract_text_from_pdf
from utils.settings import base_settings
from utils.summarize import summarize_transactions
from utils.websocket import WebSocketManager

# Replace global ws_connections with typed version
ws_connections: set[WebSocketResponse] = set()
ws_lock = Lock()


async def start_background_tasks(app):
    """Initialize application background tasks."""
    app['ws_connections'] = ws_connections
    app['ws_lock'] = ws_lock


async def cleanup_background_tasks(app):
    """Cleanup application resources."""
    await cleanup_ws(app)


async def cleanup_ws(app):
    """Cleanup WebSocket connections on shutdown."""
    async with ws_lock:
        connections = set(ws_connections)  # Create a copy to iterate safely
        for ws in connections:
            await ws.close(code=WSMsgType.CLOSE, message='Server shutdown')
        ws_connections.clear()

async def extract_text(request: Request) -> Response:
    try:
        base_settings.logger.info('Received text extraction request')
        reader = await request.multipart()
        field = await reader.next()

        if field.name != 'file':
            base_settings.logger.warning('No file field in request')
            return web.json_response({'error': 'No file uploaded'}, status=400)

        # Read the file content
        base_settings.logger.info('Reading uploaded file')
        file_content: bytes = await field.read()

        text: str = await extract_text_from_pdf(file_content)
        base_settings.logger.info('Successfully processed request')
        return web.json_response({'text': text})
    except Exception as e:
        base_settings.logger.error(
            f'Request processing failed: {str(e)}', exc_info=True
        )
        return web.json_response({'error': str(e)}, status=500)


async def websocket_handler(request: Request) -> WebSocketResponse:
    """WebSocket handler for real-time communication."""
    ws = web.WebSocketResponse()
    await ws.prepare(request)

    async with ws_lock:
        ws_connections.add(ws)
    ws_manager = WebSocketManager(ws)
    await ws_manager.prepare()

    base_settings.logger.info('WebSocket connection established')

    try:
        async for msg in ws:
            if msg.type == WSMsgType.TEXT:
                try:
                    data = msg.json()
                    if data.get('action') == 'analyze':
                        result = await analyze_transactions(
                            data.get('transactions'), ws_manager
                        )
                        await ws_manager.send_progress(
                            'Analysis complete', 1.0, 'Analysis'
                        )
                        await ws_manager.send_result(
                            result, 'Analysis', 'analysis_complete'
                        )
                    elif data.get('action') == 'summary':
                        result = await summarize_transactions(
                            data.get('transactions'), ws_manager
                        )
                        await ws_manager.send_progress(
                            'Summary complete', 1.0, 'Summarize'
                        )
                        await ws_manager.send_result(
                            result, 'Summarize', 'summary_complete'
                        )
                    else:
                        await ws_manager.send_result(
                            {'message': 'Unknown action'}, 'Error', 'error'
                        )
                except Exception as e:
                    base_settings.logger.error(f'Message processing error: {str(e)}')
                    await ws_manager.send_result({'error': str(e)}, 'Error', 'error')
            elif msg.type == WSMsgType.ERROR:
                base_settings.logger.error(f'WebSocket error: {ws.exception()}')
    finally:
        async with ws_lock:
            ws_connections.remove(ws)
        await ws.close()
        base_settings.logger.info('WebSocket connection closed')
    return ws


def init_app() -> web.Application:
    app = web.Application()
    app.router.add_post('/extract-text', extract_text)
    app.router.add_get('/ws', websocket_handler)

    # Add startup/cleanup handlers
    app.on_startup.append(start_background_tasks)
    app.on_cleanup.append(cleanup_background_tasks)

    return app


if __name__ == '__main__':
    app = init_app()
    try:
        web.run_app(
            app,
            host='0.0.0.0',
            port=int(os.environ.get('PORT', 5173)),
        )
    except KeyboardInterrupt:
        base_settings.logger.info('Received keyboard interrupt...')
    except Exception as e:
        base_settings.logger.error(f'Server error: {e}')
    finally:
        base_settings.logger.info('Server shutdown complete.')

It was a long line of code. However, looking closely, it's very straightforward. We started out registering all our WebSocket connections in a set (to prevent duplicates). We could have done something like as suggested:

...
import weakref
...

ws_connections = web.AppKey('ws_connections', weakref.WeakSet)

...

but I went for the low-level approach and controlled the startup, and async locks to ensure thread safety, cleaning up, and stuff like that. You can use the suggested approach instead. Then comes the extract_text route handler. It's the handler that receives the user's transactions PDF file, forces that file is its name (this is my preference, you can allow any file name), and then let extract_text_from_pdf do its magic in extracting texts from the file and response with the extracted text. Let's see how extract_text_from_pdf can do this.

Step 2: Utility package - `extract_text.py`

import os
import tempfile

import pytesseract
from pdf2image import convert_from_path

from .settings import base_settings as settings


async def extract_text_from_pdf(pdf_file: bytes) -> str:
    settings.logger.info('Starting PDF text extraction')
    # Create temporary file to store uploaded PDF
    with tempfile.NamedTemporaryFile(delete=False, suffix='.pdf') as tmp_pdf:
        tmp_pdf.write(pdf_file)
        tmp_path: str = tmp_pdf.name
        settings.logger.debug(f'Created temporary file: {tmp_path}')

    try:
        # Convert to image using resolution 600 dpi
        settings.logger.info('Converting PDF to images')
        pages = convert_from_path(tmp_path)
        settings.logger.debug(f'Converted {len(pages)} pages')

        # Extract text
        text_data: str = ''
        for i, page in enumerate(pages, 1):
            settings.logger.debug(f'Processing page {i}')
            text: str = pytesseract.image_to_string(page)
            text_data += text + '\n'

        settings.logger.info('Text extraction completed successfully')
        return text_data
    except Exception as e:
        settings.logger.error(f'Error during text extraction: {str(e)}', exc_info=True)
        raise
    finally:
        # Clean up temporary file
        settings.logger.debug(f'Cleaning up temporary file: {tmp_path}')
        os.unlink(tmp_path)

It uses Google OCR via the pytesseract library to extract texts from the file. Although, OCR should be able to get texts directly from any document but it worked better for me with images via image_to_string hence the intermediate step of converting the PDF file to images before feeding it into OCR. The temporary file creation was because convert_from_path requires a path that could only be gotten from a temporary file at this point since we don't want to save anyone's file on our end. A better option is to use the convert_from_bytes from pdf2image which doesn't require temporary file creation.

Back to app.py, we also have a WebSocket handler that simply delegates actions to various submodules to handle. But before then, I have a utility that helps manage all the app's WebSocket connections. This is to elegantly manage connections without worrying about leaks and stuff like that.

Step 3: Utility package - `websocket.py`

from aiohttp import web

from utils.settings import base_settings


class WebSocketManager:
    def __init__(self, ws: web.WebSocketResponse):
        self.ws = ws
        self._ready = False
        base_settings.logger.info(f'Initializing WebSocket manager: {ws}')

    async def prepare(self):
        self._ready = True
        base_settings.logger.info('WebSocket manager ready')

    async def send_progress(self, message: str, progress: float, task_type: str = None):
        if not self._ready or self.ws.closed:
            base_settings.logger.warning(
                'Cannot send progress - WebSocket not ready/closed'
            )
            return

        try:
            await self.ws.send_json(
                {
                    'action': 'progress',
                    'message': message,
                    'progress': progress,
                    'taskType': task_type,
                }
            )
        except Exception as e:
            base_settings.logger.error(f'Error sending progress: {str(e)}')

    async def send_result(self, result: dict, task_type: str, action: str):
        if not self._ready or self.ws.closed:
            base_settings.logger.warning(
                'Cannot send result - WebSocket not ready/closed'
            )
            return

        try:
            await self.ws.send_json(
                {
                    'action': action,
                    'result': result,
                    'taskType': task_type,
                }
            )
        except Exception as e:
            base_settings.logger.error(f'Error sending result: {str(e)}')

The app would well work without it but I encountered a bug where connections were not ready but requests were already being made by the frontend. So I figured having this utility would help to ensure the readiness of these connections before accepting requests. I also implemented methods, send_result and send_progress, to utilize aiohttp's WebSocket's send_json to send analysis results and analysis progress reports respectively back to the requester(s) so that they won't be kept in the dark if the analysis is taking long (which will on a CPU-only machine using many transformer models). After the custom WebSocket manager completes preparation, we loop over the messages received and appropriately delegate actions based on the action key. For now, we support analyze and summary actions via the analyze_transactions and summarize_transactions respectively. Let's see what they do.

Step 4: Utility package - `analyzer.py`

import asyncio
import os
from datetime import datetime

import numpy as np
import pandas as pd
import torch
from sklearn.ensemble import IsolationForest
from transformers import pipeline

from models.base import Transaction
from utils.settings import base_settings as settings
from utils.websocket import WebSocketManager


def get_device() -> tuple[torch.device, str]:
    """
    Detect the best available device (GPU, MPS, or CPU) for PyTorch computations.
    """
    if torch.cuda.is_available():
        # Check if CUDA (NVIDIA GPU) is available
        return torch.device('cuda'), 'CUDA (NVIDIA GPU)'
    elif torch.backends.mps.is_available():
        # Check if MPS (Metal Performance Shaders on Apple Silicon) is available
        return torch.device('mps'), 'MPS (Apple Metal)'
    else:
        # Default to CPU
        return torch.device('cpu'), 'CPU'


async def analyze_transactions(
    transactions: list[dict], ws_manager: WebSocketManager = None
) -> dict:
    """Analyze transactions and return insights with progress updates."""
    try:
        # Step 1: Validate and preprocess transactions
        if ws_manager:
            await ws_manager.send_progress(
                'Validating transactions...', 0.1, 'Analysis'
            )

        if not transactions:
            if ws_manager:
                await ws_manager.send_progress(
                    'No transactions provided', 1.0, 'Summarize'
                )
            return {'error': 'No transactions provided'}

        tx_objects = [
            Transaction(
                _id=t['_id'],
                balance=float(t['balance']),
                type=t['type'],
                date=datetime.fromisoformat(t['date']),
                description=t['description'],
                amount=float(t['amount']),
                userId=t['userId'],
                createdAt=datetime.fromisoformat(t['createdAt']),
                updatedAt=datetime.fromisoformat(t['updatedAt']),
            )
            for t in transactions
            if validate_transaction(t)
        ]

        if not tx_objects:
            if ws_manager:
                await ws_manager.send_progress(
                    'No valid transactions provided', 1.0, 'Analysis'
                )
            return {'error': 'No valid transactions provided'}

        # Step 2: Classification
        if ws_manager:
            await ws_manager.send_progress(
                'Classifying transactions...', 0.2, 'Analysis'
            )
        categories = await classify_transactions(tx_objects)

        # Step 3: Anomaly Detection
        if ws_manager:
            await ws_manager.send_progress('Detecting anomalies...', 0.4, 'Analysis')
        anomalies = await detect_anomalies(tx_objects)

        # Step 4: Spending Analysis
        if ws_manager:
            await ws_manager.send_progress('Analyzing spending...', 0.6, 'Analysis')
        spending_analysis = await analyze_spending(tx_objects)

        # Step 5: Trend Prediction
        if ws_manager:
            await ws_manager.send_progress(
                'Predicting spending trends...', 0.8, 'Analysis'
            )
        spending_trends = await predict_trends(tx_objects)

        # Compile the results
        result = {
            'categories': categories,
            'anomalies': anomalies,
            'spending_analysis': spending_analysis,
            'spending_trends': spending_trends,
        }

        settings.logger.info('Transaction analysis completed successfully')
        return result

    except Exception as e:
        settings.logger.error(f'Error analyzing transactions: {str(e)}', exc_info=True)
        if ws_manager:
            await ws_manager.send_progress('Analysis failed', 1.0)
        return {'error': f'Analysis failed: {str(e)}'}


def validate_transaction(t: dict) -> bool:
    """Validate transaction fields."""
    try:
        required_fields = {'date', 'description', 'amount', 'balance', 'type', 'userId'}
        if not all(field in t for field in required_fields):
            return False

        # Validate amount and balance are numeric
        float(t['amount'])
        float(t['balance'])

        # Validate date formats
        datetime.fromisoformat(t['date'])

        return True
    except (ValueError, TypeError):
        return False


async def classify_transactions(transactions: list[Transaction]) -> dict:
    """
    Classify transactions using FinBERT.
    """
    # Get device (GPU if available, otherwise CPU)
    device, device_name = get_device()
    settings.logger.info(f'Using device for classification: {device_name}')
    # Load FinBERT classification pipeline
    classifier = pipeline(
        'zero-shot-classification',
        model='yiyanghkust/finbert-tone',
        device=0 if device.type in ['cuda', 'mps'] else -1,  # Use GPU if available
    )

    # Define financial categories
    labels = os.getenv(
        'LABELS', 'groceries,housing,transportation,entertainment,utilities,other'
    ).split(',')

    # Prepare transaction descriptions for classification
    descriptions = [tx.description.lower() for tx in transactions]

    # Batch classify descriptions (Improves performance)
    results = await asyncio.to_thread(
        classifier, descriptions, labels, truncation=True, max_length=128
    )

    # Initialize categories and percentages
    categories = {label: 0 for label in labels}

    # Aggregate classification results
    for tx, result in zip(transactions, results):
        category = result['labels'][0]
        categories[category] += abs(tx.amount)

    total_spent = sum(categories.values())

    # Calculate percentages
    percentages = {
        category: (amount / total_spent) * 100 if total_spent > 0 else 0
        for category, amount in categories.items()
    }

    return {'categories': categories, 'percentages': percentages}


async def detect_anomalies(transactions: list[Transaction]) -> list[dict]:
    """Detect anomalies in transactions and provide specific reasons."""
    # Extract transaction amounts and reshape for Isolation Forest
    amounts = np.array([tx.amount for tx in transactions]).reshape(-1, 1)
    model = IsolationForest(contamination=0.05, random_state=42)
    anomalies = model.fit_predict(amounts)

    # Calculate mean and standard deviation for dynamic reason generation
    mean = np.mean(amounts)
    std = np.std(amounts)

    # Detect anomalies and generate specific reasons
    anomaly_details = []
    for tx, anomaly in zip(transactions, anomalies):
        if anomaly == -1:  # Anomaly detected
            # Calculate the Z-score for the transaction
            z_score = (tx.amount - mean) / std

            # Generate a dynamic reason
            if tx.amount > 0:
                reason = (
                    f'Unusually high income of {tx.amount} detected '
                    f'(Z-score: {z_score:.2f}).'
                )
            elif tx.amount < 0 and abs(tx.amount) > abs(mean) + 2 * std:
                reason = (
                    f'Unusually large expense of {tx.amount} detected '
                    f'(Z-score: {z_score:.2f}).'
                )
            elif tx.amount < 0 and 'luxury' in tx.description.lower():
                reason = 'Uncommon luxury expense detected.'
            elif tx.amount < 0 and 'groceries' in tx.description.lower():
                reason = 'Unusually high grocery expense detected.'
            else:
                reason = f'Outlier transaction with amount {tx.amount} (Z-score: {z_score:.2f}).'

            # Append the anomaly details
            anomaly_details.append(
                {
                    'date': tx.date.isoformat(),
                    'description': tx.description,
                    'amount': tx.amount,
                    'reason': reason,
                }
            )

    return anomaly_details


async def analyze_spending(transactions: list[Transaction]) -> dict:
    """Generate spending analysis with cumulative balance"""
    total_spent = sum(tx.amount for tx in transactions if tx.amount < 0)
    total_income = sum(tx.amount for tx in transactions if tx.amount > 0)

    # Create a DataFrame from transactions
    df = pd.DataFrame([t.__dict__ for t in transactions])

    # Ensure date parsing is correct
    df['date'] = pd.to_datetime(df['date'])
    df['date'] = df['date'].dt.tz_localize(None)

    # Group by the date and calculate daily totals
    daily_summary = df.groupby(df['date'].dt.date)['amount'].sum()

    # Sort by date to ensure cumulative calculations are correct
    df = df.sort_values(by='date')

    # Calculate the cumulative balance
    df['cumulative_balance'] = df['balance']

    # Convert daily_summary to JSON-serializable format
    daily_summary = {str(date): float(amount) for date, amount in daily_summary.items()}

    # Prepare cumulative balance as JSON-serializable format
    cumulative_balance = {
        row['date'].strftime('%Y-%m-%d'): row['cumulative_balance']
        for _, row in df.iterrows()
    }

    return {
        'total_spent': abs(total_spent),
        'total_income': total_income,
        'savings_rate': (
            ((total_income + total_spent) / total_income) * 100 if total_income else 0
        ),
        'daily_summary': daily_summary,
        'cumulative_balance': cumulative_balance,
    }


async def predict_trends(transactions: list[Transaction]) -> dict:
    """Predict future spending trends with enhanced analysis."""
    if len(transactions) < 2:
        return {'trend': 'Not enough data'}

    # Convert dates to numeric for regression
    dates = [(tx.date - transactions[0].date).days for tx in transactions]
    amounts = [tx.amount for tx in transactions]

    # Linear regression for trends
    coeffs = np.polyfit(dates, amounts, 1)
    trend = 'increasing' if coeffs[0] > 0 else 'decreasing'

    # Include confidence interval (optional)
    slope, intercept = coeffs
    trend_line = [slope * x + intercept for x in dates]

    # Estimated monthly spend
    df = pd.DataFrame([t.__dict__ for t in transactions])
    df['date'] = pd.to_datetime(df['date'])
    df['date'] = df['date'].dt.tz_localize(None)
    months = len(df['date'].dt.to_period('M').unique())

    return {
        'trend': trend,
        'trend_slope': slope,
        'estimated_monthly_spend': abs(
            sum(tx.amount for tx in transactions if tx.amount < 0)
        )
        / (months or 1),
    }

It's full of async functions that handle specific cases in the analysis (not minding transaction validation and device detection logic). A particular function of interest is the classify_transactions which uses yiyanghkust/finbert-tone, A Large Language Model for Extracting Information from Financial Text, to perform Zero-Shot Classification of the transactions mostly into:

LABELS=groceries,school,housing,transportation,gadgets,entertainment,utilities,credit cards,miscellaneous,dining out,healthcare,insurance,savings,investments,childcare,travel,personal care,debts,charity,taxes,subscriptions,streaming services,home maintenance,shopping,pets,fitness,hobbies,gifts

Realistically, it'd be more accurate to finetune a barebone Bidirectional Encoder Representations from Transformers (BERT) such as RoBERTa, DistilBERT, and co with our data but my excuse was lack of adequate data and GPU (Google Colab isn't enough).

Tip: A nice way to collaborate

If you want to explore the idea behind finetuning and barebone Natural Language Processing with transformers, we can collaborate to create something like FinBERT but for account statements and co. Please reach out.

There are other functions for detecting incomes/expenses that are "abnormal" by using IsolationForest and Z-score for the detection and for generating human-readable reasons for such a detection. Cool stuff!!! There were also functions for spending analysis and stuff but they are pretty basic.

Note: amount assumption

All the analysis here assumed that income has a positive amount while expenses possess negative amounts. If your data is different from this assumption, you may need to modify the code or data depending on your preference.

Before we roundup, let's see what utils/summarize.py is

Step 5: Utility package - `summarize.py`

from datetime import datetime

import pandas as pd

from models.base import Transaction
from utils.analyzer import validate_transaction
from utils.settings import base_settings as settings
from utils.websocket import WebSocketManager


async def summarize_transactions(
    transactions: list[dict], ws_manager: WebSocketManager = None
) -> dict:
    """Summarize transaction data."""
    try:
        # Validate and convert transactions to objects
        if ws_manager:
            await ws_manager.send_progress(
                'Validating transactions...', 0.1, 'Summarize'
            )

        if not transactions:
            if ws_manager:
                await ws_manager.send_progress(
                    'No transactions provided', 1.0, 'Summarize'
                )
            return {'error': 'No transactions provided'}

        tx_objects = [
            Transaction(
                _id=t['_id'],
                balance=float(t['balance']),
                type=t['type'],
                date=datetime.fromisoformat(t['date']),
                description=t['description'],
                amount=float(t['amount']),
                userId=t['userId'],
                createdAt=datetime.fromisoformat(t['createdAt']),
                updatedAt=datetime.fromisoformat(t['updatedAt']),
            )
            for t in transactions
            if validate_transaction(t)
        ]

        if not tx_objects:
            settings.logger.warning('No valid transactions provided')
            if ws_manager:
                await ws_manager.send_progress(
                    'No valid transactions provided', 1.0, 'Summarize'
                )
            return {'error': 'No valid transactions provided'}

        # Step 1: Calculate totals
        if ws_manager:
            await ws_manager.send_progress('Calculating totals...', 0.2, 'Summarize')

        total_spent = sum(tx.amount for tx in tx_objects if tx.amount < 0)
        total_income = sum(tx.amount for tx in tx_objects if tx.amount > 0)

        # Step 2: Calculate additional metrics
        if ws_manager:
            await ws_manager.send_progress('Calculating average..', 0.35, 'Summarize')

        total_savings = total_income + total_spent
        total_transactions = len(tx_objects)
        expense_count = sum(1 for tx in tx_objects if tx.amount < 0)
        income_count = sum(1 for tx in tx_objects if tx.amount > 0)
        avg_expense = abs(total_spent / expense_count) if expense_count > 0 else 0
        avg_income = total_income / income_count if income_count > 0 else 0

        # Step 3: Calculate largest transactions and date range
        if ws_manager:
            await ws_manager.send_progress(
                'Identifying largest transactions...', 0.5, 'Summarize'
            )

        start_date = min(tx.date for tx in tx_objects)
        end_date = max(tx.date for tx in tx_objects)
        largest_expense = min(tx.amount for tx in tx_objects if tx.amount < 0)
        largest_income = max(tx.amount for tx in tx_objects if tx.amount > 0)

        # Step 4: Generate monthly summaries
        if ws_manager:
            await ws_manager.send_progress(
                'Generating monthly summaries...', 0.7, 'Summarize'
            )

        df = pd.DataFrame([t.__dict__ for t in tx_objects])
        df['date'] = pd.to_datetime(df['date'])
        monthly_summary = (
            df.groupby(df['date'].dt.to_period('M'))['amount'].sum().to_dict()
        )

        # Step 5: Analyze trends and changes
        if ws_manager:
            await ws_manager.send_progress('Analyzing trends...', 0.85, 'Summarize')

        monthly_income = (
            df[df['amount'] > 0].groupby(df['date'].dt.to_period('M'))['amount'].sum()
        )
        monthly_expense = (
            df[df['amount'] < 0]
            .groupby(df['date'].dt.to_period('M'))['amount']
            .sum()
            .abs()
        )
        monthly_savings = (monthly_income - monthly_expense).fillna(0)

        income_trend = await calculate_trend(monthly_income)
        expense_trend = await calculate_trend(monthly_expense)
        savings_trend = await calculate_trend(monthly_savings)

        income_change = await calculate_percentage_change(monthly_income)
        expense_change = await calculate_percentage_change(monthly_expense)
        savings_change = await calculate_percentage_change(monthly_savings)

        savings_rate = (
            ((total_income + total_spent) / total_income) * 100 if total_income else 0
        )

        monthly_summary = {
            str(month): {
                'income': float(monthly_income.get(month, 0)),
                'expenses': float(monthly_expense.get(month, 0)),
                'savings': float(monthly_savings.get(month, 0)),
            }
            for month in monthly_income.index.union(monthly_expense.index)
        }

        # Compile summary results
        summary = {
            'income': {
                'total': total_income,
                'trend': income_trend,
                'change': income_change,
            },
            'expenses': {
                'total': abs(total_spent),
                'trend': expense_trend,
                'change': expense_change,
            },
            'savings': {
                'total': total_savings,
                'trend': savings_trend,
                'change': savings_change,
            },
            'total_transactions': total_transactions,
            'expense_count': expense_count,
            'income_count': income_count,
            'avg_expense': avg_expense,
            'avg_income': avg_income,
            'start_date': start_date.isoformat(),
            'end_date': end_date.isoformat(),
            'largest_expense': largest_expense,
            'largest_income': largest_income,
            'savings_rate': savings_rate,
            'monthly_summary': monthly_summary,
        }
        settings.logger.info('Transaction summarization completed successfully')
        return summary
    except Exception as e:
        settings.logger.error(f'Error summarizing transactions: {str(e)}')
        if ws_manager:
            await ws_manager.send_progress('Summarization failed', 1.0)
        return {'error': f'Summarization failed: {str(e)}'}


async def calculate_trend(monthly_data: pd.Series) -> str:
    """
    Calculate trend ('up', 'down', 'neutral') based on monthly data.
    """
    if len(monthly_data) < 2:
        return 'neutral'

    recent_avg = monthly_data[-2:].mean()
    earlier_avg = monthly_data[:-2].mean()

    if recent_avg > earlier_avg:
        return 'up'
    elif recent_avg < earlier_avg:
        return 'down'
    else:
        return 'neutral'


async def calculate_percentage_change(monthly_data: pd.Series) -> float:
    """
    Calculate the percentage change from the highest monthly value to the average of the last two months.
    """
    if len(monthly_data) < 2:
        return 0

    highest_value = monthly_data.max()
    recent_avg = monthly_data[-2:].mean()

    return (
        ((recent_avg - highest_value) / highest_value) * 100
        if highest_value != 0
        else 0
    )

Though long (due to repeated codes that could have been drafted into reusable functions), the function just does basic summaries of your transactions. It's comprehensive but can be well extended.

In both analyses, I endeavored to update connections with progress reports by sending progress actions at intervals informing the user of the time left.

Now, in app.py, using init_app, we created an Application instance and added routes to it. We could have used @route decorator instead. We didn't forget to do some housekeeping (clean-ups) of the background tasks previously started whenever we quit the app via different interrupts. That's it!

Note: Undiscussed utilities

There were some submodules: utils/settings.py, models/base.py, etc, not discussed. They are basic setups for instrumentation (logging) and stuff. Others are for another system (maybe discussed in another series). In all, they are very basic.

Outro

If you found this article valuable, consider sharing it with your network to help spread the knowledge!

Building an AI-powered Financial Behavior Analyzer with NodeJS, Python, SvelteKit, and TailwindCSS - Part 0

John Owolabi Idogun — Fri, 31 Jan 2025 07:41:07 +0000

Introduction

In the ever-evolving landscape of Software Engineering, staying up-to-date with the latest technologies is crucial. Earlier this year, I wanted to refresh my skills with Node.js (Express.js), having been a while since I last delved into them. Since I learn better by doing, I needed a project to work on but I didn't just want to build another run-of-the-mill application. I wanted a project that would be both challenging and exciting, something that would allow me to explore the fascinating world of AI while keeping my hands on the core development. Then the idea of creating an AI-powered financial data analyzer sparked my interest. It presented the perfect opportunity to not only revisit Node and its ecosystem but also to dive into the realm of AI/ML by working with transformers and lower-level libraries like PyTorch and Hugging Face, rather than relying on pre-built, heavy AI APIs. Thus, the AI Financial Analyzer project was born. This series of articles will walk you through the architecture and implementation of this project, showcasing how you can leverage SvelteKit (or any frontend framework), Tailwind CSS (v4), Express.js, and Python's AI ecosystem to build a robust and insightful financial analysis tool.

Source code

Sirneij / finance-analyzer

An AI-powered financial behavior analyzer and advisor written in Python (aiohttp) and TypeScript (ExpressJS & SvelteKit with Svelte 5)

View on GitHub

System architecture and requirements

In this article, I’ll walk you through the architecture of the system we'll be building. In subsequent articles, we will delve into implementation details. So, let's explore the entire architecture, from frontend to backend, and how AI-powered transaction analysis works.

Architecture Overview

The system consists of three main components:

SvelteKit Frontend (Browser)
Node.js Backend Server (Middleman)
AI Service (Python-based Analysis)

Component 1: SvelteKit Frontend (User Interface)

The frontend will be built with SvelteKit (Svelte 5), leveraging TailwindCSS v4 for styling and Chart.js for interactive financial data visualization. It will interact with the backend via REST APIs and WebSockets.

Frontend Requirements

User Authentication: Utilizes GitHub OAuth for user sign-in (Google OAuth planned).
Data Input: Should allow users to upload financial data (CSV, PDF, and Excel — Excel support will be considered) or manually input transactions.
Financial Dashboard: Displays real-time financial analytics, including income vs. expenses, savings rate, and other insights.
WebSockets Support: Ensures real-time updates when AI analysis results are received from the backend.

Component 2: Node.js Backend (Express.js & MongoDB)

The backend, built with Node.js (Express.js), will act as the core application logic layer. It'll handle authentication, database management, and communication between the frontend and AI service.

Backend Requirements

User Authentication & Database Management:
- OAuth authentication (GitHub supported, Google planned).
- Stores user-profiles and transactions in MongoDB.
Data Parsing & API Handling:
- Provides REST APIs to parse transaction data from CSV, PDF, and Excel.
- Saves parsed transactions in the database for later analysis.
WebSockets for AI Communication:
- Acts as a middleman between the frontend and AI service.
- Sends user-specific transaction data to the AI service.
- Filters and authenticates user data before analysis.
- Returns analyzed results via WebSockets for real-time updates.

Component 3: AI Service (Python)

The AI Service will be built using aiohttp (asynchronous Python web server) and integrates PyTorch, Hugging Face Transformers, numpy, pandas, and scikit-learn for financial data analysis.

AI Service Requirements

PDF Transaction Parsing:
- Uses pdf2image and pytesseract for OCR-based extraction of transaction data.
- Sends extracted data back to Node.js for validation and storage.
Transaction Analysis & Categorization:
- Uses PyTorch & Transformers for AI-driven categorization (Income, Expenses, Savings, etc.).
- Computes total transactions, savings rate, and spending trends.
- Results are streamed via WebSockets back to the frontend for real-time updates.

We'll stop here for now. You are encouraged to set up your machine for the development:

[ ] For NodeJS, ensure you have v20+. Set it up for an ExpressJs using TypeScript project.
[ ] Get SvelteKit (with Svelte 5) setup and include TailwindCSS and TypeScript.
[ ] Create an environment variable and install those Python tools. PyTorch may take a while to conclude.

These will be assumed in the next article.

Outro

If you found this article valuable, consider sharing it with your network to help spread the knowledge!

TailwindCSS v4.0: Upgrading from v3 with some plugins

John Owolabi Idogun — Fri, 31 Jan 2025 01:50:10 +0000

Introduction

Recently, while building a personal blog started with Tailwind CSS v3.4, I was faced with the need to upgrade to the latest version of the utility-first library. Though this migration guide was helpful, I couldn't find resources to include plugins such as @tailwindcss/typography (needed for the prose-related rules) and @tailwindcss/forms (needed for form elements). Hence this article. It will be a short ride.

Prerequisite

It is assumed you have a basic familiarity with the library. Also, I will be using SvelteKit as a framework of choice here but you just need any project where you need to migrate tailwind CSS from v3 to v4.

Implementation

Step 1: Run the tailwind CSS upgrade tool

To ease developers' stress migrating, the team at Tailwind CSS provided a nifty tool that helps to effortlessly migrate between the versions. In your project's root, just run:

$ npx @tailwindcss/upgrade@next

You should accept the installation prompt and watch it do its magic! It will help correct some breaking changes in your code, fix postcss configuration (if you use it), correct the import statements in your main CSS file where tailwind was used, upgrade its version in package.json, etc. In case it can't migrate some stuff, you will be informed.

Step 2: Using plugins

The step above was pretty easy and basic. As I mentioned, the migration guide included that. However, most projects use more than just the basic tailwind setup. For example, you need @tailwindcss/typography for this rule to work:

.content {
  @apply dark:prose-invert prose-headings:font-bold prose-a:no-underline hover:prose-a:underline prose-pre:rounded-lg prose-pre:p-4;
}

without it, you will be greeted with something like:

Cannot apply unknown utility class: dark:prose-invert

In the same vein, without @tailwindcss/forms, you would find it hard to easily work with form elements, especially checkboxes.

In version 3, adding plugins such as the above would be done in tailwind.config.ts:

...
import forms from '@tailwindcss/forms';
import typography from '@tailwindcss/typography';
import type { Config } from 'tailwindcss';

export default {
    ...
    theme: {
        ...
    },
    plugins: [typography, forms,...]
} satisfies Config;

However, with v4, you can ditch that file entirely. Instead, use your main entry CSS file (where you import tailwindcss):

@import "tailwindcss";
@plugin "@tailwindcss/typography";
@plugin "@tailwindcss/forms";
@plugin ...;

with that, tailwindcss will compile easily.

Step 3: Enabling dark mode

By default, v4 uses the awesome prefers-color-scheme media feature to intrinsically support dark mode. However, most of us are used to toggling light/dark themes based on classes. Hence, in v3, we achieved this using tailwind.config.ts:

...
export default {
    ...
    darkMode: 'class',
    theme: {
        ...
    },
    plugins: [...]
} satisfies Config;

In v4, the file will also be ditched. The main entry CSS file comes to the rescue:

@import "tailwindcss";
...

@custom-variant dark (&:where(.dark, .dark *));;

with that, your previous toggling feature comes back to life!

That's it! If you have other tips, don't hesitate to share them.

Outro

If you found this article valuable, consider sharing it with your network to help spread the knowledge!

DEV Community: John Owolabi Idogun

Integrating Stripe with SvelteKit: Dynamic Pricing and Payment Security

Introduction

Live version

Source code

Sirneij / digibooks

A simple demo app that integrates Stripe with SvelteKit

Project Specification: Simple Digital Books App with SvelteKit & Stripe

1. Project Goal

2. Core Features

2.1. Book Catalog

2.2. Shopping Cart

2.3. Stripe Checkout

System architecture and requirements

Prerequisite

Implementation

Step 1: Create the database schema

Seeding the database

Step 2: Cart and carting store

Step 3: Data loading and frontend

Step 4: Payment integration

Step 5: Confirming purchases with webhook

Outro

Quantum Information Processing: Foundations - Part 1

Introduction

Prerequisite

Dirac notation

Superposition Principle and Measurements

Question 1:

Solution:

Check:

Code implementation using Google cirq

Question 2:

Solution

Check:

Question 3:

Solution

Check

Question 4:

Solution

Outro

How to Build a Dual-LLM Chat Application with Next.js, Python, and WebSocket Streaming

Introduction

Prerequisite

Source code

Sirneij / genai-chatbot

A streaming chatbot powered by Large Language Models

GenAI ChatBot

Overview

Quick Start

Features

Project Structure

Implementation

Step 1: Theory - Autoregressive vs Autoencoding models

Step 2: Python asynchronous backend with aiohttp

Step 3: Building a React Frontend with Next.js and Tailwind CSS

Outro

Asynchronous Server: Building and Rigorously Testing a WebSocket and HTTP Server

Introduction

Prerequisite

Source code

Sirneij / finance-analyzer

An AI-powered financial behavior analyzer and advisor written in Python (aiohttp) and TypeScript (ExpressJS & SvelteKit with Svelte 5)

Implementation

Step 1: Improving the Initial AI Service

Step 2: Enforcing Type Safety, Security, and Style

Step 3: Writing the tests

Outro

Building an AI-powered Financial Behavior Analyzer with NodeJS, Python, SvelteKit, and TailwindCSS - Part 4

Introduction

Prerequisite

Source code

Sirneij / finance-analyzer

An AI-powered financial behavior analyzer and advisor written in Python (aiohttp) and TypeScript (ExpressJS & SvelteKit with Svelte 5)

Implementation

Step: WebSocket handler

Step 2: SvelteKit with TailwindCSS

Step 3: Leverage prefers-color-scheme in theme toggling

Step 4: Theme switching logic with icons

Step 5: Authentication or login page

Code implementation using Google `cirq`

Step 3: Leverage `prefers-color-scheme` in theme toggling

NOTE: If you use Coingecko's API, when you use my code, `CGSIRNEIJ`, I get some commissions. That can be a good way to help me.

Step 2: Define `FullCoin` model

Step 3: Business logic in `transaction.service.ts`

Step 3: Setting up an express server in `src/app.ts`

Step 1: Spawning an `aiohttp` Server

Step 2: Utility package - `extract_text.py`

Step 3: Utility package - `websocket.py`

Step 4: Utility package - `analyzer.py`