Forem: Ivan Korostenskij

Why Your Repository Pattern Creates Tech Debt (And How to Fix It in Python)

Ivan Korostenskij — Thu, 02 Apr 2026 02:11:06 +0000

The Repository Pattern in Python

We've all inherited it: a critical 500-line function with raw SQL strings precariously placed between error handling, business logic, and an API call. You feel a natural instinct to refactor it, separate concerns; that's the right call! However, the pattern most tutorials teach you to accomplish this just creates a different kind of mess.

I'm talking about the repository pattern: an approach to separate your business layer (business logic) from your data layer (persistence and retrieval from a database).

Understanding what drives this pattern - and where the standard implementation goes wrong - will permanently change how you structure database logic. The end result is maintainable, testable, and readable code.

Martin Fowler describes this as mediating interaction, "… between the domain and data mapping layers using a collection-like interface." Let's break down what that actually means, and how it's been so misinterpreted.

The (short) theory

The fundamentals are exactly the same between good and bad approaches.

1) The "collections-like interface" is this: a class that defines database method signatures. In Python, we use Protocol from the typing module - this gives us structural subtyping (essentially implicit interfaces, like Go):

from typing import Protocol
from datetime import datetime

class UserRepository(Protocol):
    async def get_user_birthday_by_id(self, user_id: int) -> datetime: ...
    async def create_user(self, user: User) -> None: ...
    async def delete_user(self, user: User) -> None: ...

2) The implementation is a class with a single Unit of Work (database session) attribute.

Unit of Work

I highly recommend this article for a deep dive on the Unit of Work in Python

Ivan Korostenskij

Dec 11 '25

Stop Writing try/except Hell: Clean Database Transactions with SQLAlchemy with the Unit Of Work

#python #architecture #tutorial #learning

Comments 4

5 min read

from sqlalchemy import select
from unit_of_work import UnitOfWork

class UserStore:  # implements UserRepository
    def __init__(self, uow: UnitOfWork) -> None:
        self._uow = uow

    async def get_user_birthday_by_id(self, user_id: int) -> datetime:
        result = await self._uow.session.execute(
            select(User.birth_date).where(User.id == user_id)
        )
        return result.scalar_one()  # raises NoResultFound if missing

    async def create_user(self, user: User) -> None:
        self._uow.session.add(user)

    async def delete_user(self, user: User) -> None:
        await self._uow.session.delete(user)

Notice: UserStore never inherits from UserRepository. Python's Protocol uses structural subtyping - if the class has the right methods with the right signatures, it satisfies the protocol automatically. No inheritance required. This is the same behavior as Go's implicit interface satisfaction.

A single decision determines whether this approach becomes pure tech debt or maintainable, readable code: where you put it.

The Bad Way

To create a code version of the Atlantic garbage patch, follow outdated tutorials and define a single, giant protocol of database methods per table.

# Lives in: src/everything/user_repository_and_motorcycle_parts_slash_comic_book_store.py

class UserRepository(Protocol):
    async def get_user_birthday_by_id(self, user_id: int) -> datetime: ...
    async def create_user(self, user: User) -> None: ...
    async def delete_user(self, user: User) -> None: ...
    # ... 50 more methods for every niche edge case

This now serves as mini dumping grounds of every method any service needs, across your application.

This is a producer-defined interface - the repository declares everything it can do, and every consumer must accept the whole thing. Any test that touches this needs to mock all 50+ methods.

Let's see how this approach scales, starting fresh.

Two engineers - Sarah and Mike - start developing separate features, working with user data. Sarah needs an add_or_upgrade_user_subscription_tier database method to upgrade paying users (or add them if they're on a Free account).

class UserRepository(Protocol):
    async def add_or_upgrade_user_subscription_tier(
        self, user_id: int, tier: Tier
    ) -> None: ...

Mike now needs a method that adds time, in days, to a user's subscription. It's somewhat related, but he doesn't have a choice where to put it - it goes in the hole.

He can either:

1) Widen the existing method: Expand Sarah's query to accommodate his work, forcing all of its callers to follow the unrelated contract he shoved into it

class UserRepository(Protocol):
    async def add_or_upgrade_user_subscription_tier_or_time(
        self, user_id: int, tier: Tier | None = None, extend_days: int | None = None
    ) -> None: ...

2) Add a near-duplicate method: Break the interface segregation principle so every service working with users now has +1 extra useless method, adding a single-use, near-identical method

class UserRepository(Protocol):
    async def add_or_upgrade_user_subscription_tier(
        self, user_id: int, tier: Tier
    ) -> None: ...
    async def add_or_upgrade_user_subscription_time(
        self, user_id: int, extend_days: int
    ) -> None: ...

Neither are good. The interface grows either way and becomes a Pandora's box of hundreds of tangentially related queries. The implementation of this interface is even worse: easily breaking thousands of lines of unoptimized ORM code in a single file as the unavoidable blast radius of each subsequent change climbs.

"The bigger the interface, the weaker the abstraction."

Eventually, we have a god object that is injected into all parts of your code that have to touch user data. Testing your single method becomes a game of ensuring the other 50+ are mocked.

So, how can solely changing the location of these methods transform this approach into the gold standard for maintainable database logic? By spreading it back out.

The good way

Martin Fowler didn't say the "collection-like interfaces" need to be Python files that are thousands of lines long. So let's make them smaller and more focused. Instead of defining a 1000-point Swiss army knife for your application, we let each service define exactly what it needs: just a screwdriver; a hammer and 3 nails; a butter knife and some tweezers.

We switch from producer-defined interfaces to consumer-defined interfaces, making specialized, mini-repositories per feature.

In Python, the Service defines the Protocol, and the Repository satisfies it structurally - without inheritance.

# src/features/notifications/service.py
from typing import Protocol
from datetime import datetime, timedelta

class NotificationStore(Protocol):
    """Service-owned protocol: only the db methods this feature needs."""
    async def get_last_notified(self, user_id: int) -> datetime: ...
    async def mark_as_notified(self, user_id: int) -> None: ...

class NotificationService(Protocol):
    async def notify_user_by_id(self, user_id: int) -> None: ...

class Service:
    def __init__(
        self, store: NotificationStore, notifier: NotificationService
    ) -> None:
        self._store = store
        self._notifier = notifier

    async def notify_user(self, user_id: int) -> None:
        last = await self._store.get_last_notified(user_id)

        if (datetime.utcnow() - last) < timedelta(hours=24):
            return

        await self._notifier.notify_user_by_id(user_id)
        await self._store.mark_as_notified(user_id)

The implementation lives in the same feature folder, satisfying the protocol of the service.

# src/features/notifications/notification_database.py
from datetime import datetime
from sqlalchemy import select, update
from unit_of_work import UnitOfWork

class PostgresStore:
    """Satisfies NotificationStore without inheriting from it."""

    def __init__(self, uow: UnitOfWork) -> None:
        self._uow = uow

    async def get_last_notified(self, user_id: int) -> datetime:
        result = await self._uow.session.execute(
            select(UserModel.last_notified_at).where(UserModel.id == user_id)
        )
        return result.scalar_one()  # raises NoResultFound if missing

    async def mark_as_notified(self, user_id: int) -> None:
        await self._uow.session.execute(
            update(UserModel)
            .where(UserModel.id == user_id)
            .values(last_notified_at=datetime.utcnow())
        )

This gives us clean, modular interfaces; readable and maintainable; testability; and a great separation of concerns between features - at the cost of a Protocol.

Figure 1: A comparison of how services connect to their separated database methods in the bad (left) vs good (right) examples

The drawback of this approach is repetition. If 5 services need a get_user() method, are we going to implement it 5 times?

Let's look at the nuances of this approach, how those 5 services probably aren't using same get_user() method you think they are, and how, "a little copying is better than a little dependency" - Go Proverbs.

Nuance (with pushback)

A different get_user() method per feature seems crazy - I'm with you; what happened to Don't Repeat Yourself (DRY)?

Upfront: application-wide repositories are okay and can absolutely be the right move. But they are often not needed.

Let's see who's calling this get_user() method:

Authentication service
Notification service
Payment service

Each of these callers needs different parts of a user's data for radically different purposes; a user in the context of billing is fundamentally a different entity than a user in the context of authentication.

A) The authentication service wants the user's username & password
B) the payment service just needs the user's payment info
C) the notification service only needs an email and the last time they were notified.

Making a global get_user() method that returns a god User object that has these + 50 more attributes - just to satisfy all callers - sounds eerily similar to the interface explosion problem we just solved.

@dataclass
class User:
    id: int
    username: str
    email: str
    password_hash: str

    plan_id: str
    subscription_status: str
    stripe_customer_id: str
    trial_ends_at: datetime | None

    last_login_at: datetime | None
    ...
    # 50 + more

Now every test that touches user data has to construct this entire object, even if the feature only cares about two fields.

I challenge you to always start local:

# src/features/notification_service/repository.py
from typing import Protocol
from dataclasses import dataclass
from datetime import datetime

class MikesNotificationStore(Protocol):
    async def get_user_last_utc_time_notified(
        self, user_id: int
    ) -> datetime: ...
    async def get_all_premium_users_with_pending_notifications(
        self,
    ) -> list[NotificationUser]: ...

# A user specialized for the service
@dataclass(frozen=True, slots=True)
class NotificationUser:
    id: int
    contact_method: str
    contact_address: str
    last_notified_at: datetime
    has_pending_notif: bool

Allow queries to be "repeated". Forcing generic CRUD methods to work with 5 disparate features increases bad coupling, maintenance cost, and lines of code - as the User object blows up to accommodate working with everything. Specific, business-case queries are the way to go.

Lean into high-affinity coupling: components that change together belong together in the same package, not sparsely connected by a generic method. Individual implementations should be able to naturally evolve with their service - at little risk to the rest of the application. Requirements will change; the question is whether your architecture will fight or accommodate that.

A note on Go

Python's Protocol and Go's implicit interfaces work the same way conceptually: the consumer defines what it needs, and any struct/class that has the right methods satisfies it - no inheritance or explicit declaration required.

I originally wrote this article for Go. If you work in Go or want to see how this pattern looks with Go's type system, check it out here:

Ivan Korostenskij

Apr 1

The Repository Pattern Done Right: Consumer-Defined Interfaces in Go

#go #architecture #beginners #learning

Comments 1

6 min read

Conclusion

Bad code is rarely a skill issue; it's a pattern issue. The wrong abstraction taught confidently in a tutorial does more damage than no abstraction at all. Knowing the benefits of the right solution matters just as much as knowing the pitfalls of the wrong one; you now have a solid grasp on both.

It wasn't Mike that caused a pileup with his addition of a method, it was starting with the wrong pattern. Bad decisions compound. So set the standard: make a consumer-defined interface today so you aren't fighting someone else's 3000-line producer-defined interface six months from now.

Next time you're adding a database method to a shared repository, setting up a new feature, or want to refactor that 500-line Jenga block, pause for a second. Ask yourself: "does every function in my codebase need launch_user_into_space_query(), or just the one I'm working with?"

I'm Ivan, follow me for more content like this! :)

The Repository Pattern Done Right: Consumer-Defined Interfaces in Go

Ivan Korostenskij — Wed, 01 Apr 2026 05:02:30 +0000

The Repository Pattern

We’ve all inherited it: a critical 500-line function with SQL Jenga blocks precariously placed between error handling, business logic, and an API call. You feel a natural instinct to refactor it, separate concerns; that’s the right call! However, the pattern most tutorials teach you to accomplish this just creates a different kind of mess.

I’m talking about the repository pattern: an approach to separate your business layer (business logic) from your data layer (persistence and retrieval from a database).

Martin Fowler describes this as mediating interaction, “… between the domain and data mapping layers using a collection-like interface.” Let’s break down what that actually means, and how it’s been so misinterpreted.

The (short) theory

The fundamentals are exactly the same between good and bad approaches.

1) The “collections-like interface” is this: an interface that defines database methods:

type UserRepository interface {
    GetUserBirthdayByID(ctx context.Context, userId uint) (*time.Time, error)
    CreateUser(ctx context.Context, p *User) error
    DeleteUser(ctx context.Context, p *User) error
    ...
}

2) The implementation is a struct with a single Unit of Work (database session) attribute.

Context: Further Reading

I highly recommend this article for a deep dive on the Unit of Work

Ivan Korostenskij

Dec 11 '25

Stop Writing try/except Hell: Clean Database Transactions with SQLAlchemy with the Unit Of Work

#python #architecture #tutorial #learning

Comments 4

5 min read

type UserStore struct { // implements UserRepository
   uow *persistence.UnitofWork
}
func (us *UserStore) GetUserBirthdayByID(ctx context.Context, userId uint) (*time.Time, error) {
        var bDay time.Time
        err := us.uow.Session.WithContext(ctx).Table("users").
        Select("id", "birth_date").Where("id = ?", userId).
        Scan(&bDay).Error
    return bDay, err
}
func (us *UserStore) CreateUser(ctx context.Context, p *User) error {
        ...
}
func (us *UserStore) DeleteUser(ctx context.Context, p *User) error {
        ...
}

A single decision decides whether this approach becomes pure tech debt or maintainable, readable code: where you put it.

The Bad Way

To create a code version of the Atlantic garbage patch, follow outdated tutorials and define a single, giant interface of database methods per table.

// Lives in: src/everything/anything_to_do_with_users_repository_and_motorcycle_parts_slash_comic_book_store.go

type UserRepository interface {
    GetUserBirthdayByID(ctx context.Context, userId uint) (*time.Time, error)
    CreateUser(ctx context.Context, p *User) error
    DeleteUser(ctx context.Context, p *User) error
    // ... 50 more methods for every niche edge case
}

This now serves as mini dumping grounds of every method any service needs, across your application.

This is a producer-defined interface — the repository declares everything it can do, and every consumer must accept the whole thing. Any test that touches this needs to mock all 50+ methods.

Let’s see how this approach scales, starting fresh.

Two engineers - Sarah and Mike - start developing separate features, working with user data. Sarah needs a AddOrUpgradeUserSubscriptionTier database method to upgrade paying users (or add them if they’re on a Free account).

type UserRepository interface {
    AddOrUpgradeUserSubscriptionTier(ctx context.Context, userId uint, t Tier) error
}

Mike now needs a method that adds time, in days, to a user’s subscription. It’s somewhat related, but he doesn’t have a choice where to put it - it goes in the hole.

He can either:

1) Widen the existing method: Expand Sarah’s query to accommodate his work, forcing all of its callers to follow the unrelated contract he shoved into it

type UserRepository interface {
    AddOrUpgradeUserSubscriptionTierOrTime(ctx context.Context, userId uint, t *Tier, t *time.Time) error
}

2) Add a near-duplicate method: Break the interface segregation principle so every service working with users now has +1 extra useless method, adding a single-use, near-identical method

type UserRepository interface {
    AddOrUpgradeUserSubscriptionTier(ctx context.Context, userId uint, t Tier) error
    AddOrUpgradeUserSubscriptionTimer(ctx context.Context, userId uint, t *time.Time) error
}

Neither are good. The interface grows either way and becomes a Pandora’s box of hundreds of tangentially related queries. The implementation of this interface is even worse: easily breaking 10k+ lines of unoptimized SQL in a single file as the unavoidable blast radius of each subsequent change climbs.

“The bigger the interface, the weaker the abstraction."

Eventually, we have a god object that is injected into all parts of your code that have to touch user data. Testing your single method becomes a game of ensuring the other 50+ are mocked.

So, how can solely changing the location of these methods transform this approach into the gold standard for maintainable database logic? By spreading it back out.

The good way

Martin Fowler didn’t say the, “collection-like interface[s]” need to be Go files that are thousands of lines long. So let’s make them smaller and more focused. Instead of defining a 1000-point Swiss army knife for your application, we let each service define exactly what it needs: just a screwdriver; a hammer and 3 nails; a butter knife and some tweezers.

We switch from producer-defined interfaces to consumer-defined interfaces, making specialized, mini-repositories per feature.

In Go, the Service defines the interface, and the Repository satisfies it.

// src/features/notifications/service.go
package notifications

// service-owned interface (declaration of the db methods it needs, modular)
type NotificationStore interface {
    GetLastNotified(ctx context.Context, userId uint) (time.Time, error)
    MarkAsNotified(ctx context.Context, userId uint) error
}

type Service struct {
    store NotificationStore
    service NotificationService
}

func (s *Service) NotifyUser(ctx context.Context, userId uint) error {
    last, err := s.store.GetLastNotified(ctx, userId)
    if err != nil { return err }

    if time.Since(last) < 24*time.Hour {
        return nil
    }
    if err := s.service.NotifyUserById(userId); err != nil {
        return err
    }

    return s.store.MarkAsNotified(ctx, userId)
}

The implementation lives in the same feature folder, satisfying the interface of the service.

// src/features/notifications/notification_database.go
package notifications

type PostgresStore struct {
   uow *persistence.UnitofWork
}

func (ps *PostgresStore) GetLastNotified(ctx context.Context, userId uint) (time.Time, error) {
    ...
}

func (ps *PostgresStore) MarkAsNotified(ctx context.Context, userId uint) error {
        ...
}

This gives us clean, modular interfaces; readable and maintainable; testability; and a great separation of concerns between features - at the cost of an interface.

The drawback of this approach is repetition. If 5 services need a GetUser() method, are we going to implement it 5 times?

Let’s look at the nuances of this approach, how those 5 services probably aren’t using same GetUser() method you think they are, and how, “a little copying is better than a little dependency” - Go Proverbs.

Nuance (with pushback)

A different GetUser() method per feature seems crazy - I’m with you; what happened to Don’t Repeat Yourself (DRY)?

Upfront: application-wide repositories are okay and can absolutely be the right move. But they are often not needed.

Let’s see who’s calling this GetUser() method:

Authentication service
Notification service
Payment service

Each of these callers needs different parts of a user’s data for radically different purposes; a user in the context of billing is fundamentally a different entity than a user in the context of authentication.

A) The authentication service wants the user’s username & password; B) the payment service just needs the user’s payment info; C) the notification service only needs an email and the last time they were notified.

Making a global GetUser() method that returns a god User object that has these + 50 more attributes - just to satisfy all callers - sounds eerily similar to the interface explosion problem we just solved.

type User struct {
    ID                int64
    Username          string
    Email             string
    PasswordHash      string 

    PlanID            string
    SubscriptionStatus string
    StripeCustomerID  string
    TrialEndsAt       *time.Time

    LastLoginAt       *time.Time
    ...
    // 50 + more
}

Now every test that touches user data has to construct this entire struct, even if the feature only cares about two fields.

I challenge you to always start local:

// src/features/notification_service/repository.go
type MikesNotificationService interface {
    GetUserLastUTCTimeNotified(ctx context.Context, userId uint) (time.Time, error)
    GetAllPremiumUsersWithPendingNotifications(ctx context.Context, userId uint) ([]NotificationUser, error)
}

// A user specialized for the service
type NotificationUser struct {
    ID                int64
  ContactMethod     string
  ContactAddress    string
    LastNotifiedAt    *time.Time
    HasPendingNotif   bool
    ...
}

Allow queries to be “repeated”. Forcing generic CRUD methods to work with 5 disparate features increases bad coupling, maintenance cost, and lines of code - as the User object blows up to accommodate working with everything. Specific, business-case queries are the way to go.

Conclusion

Bad code is rarely a skill issue; it’s a pattern issue. The wrong abstraction taught confidently in a tutorial does more damage than no abstraction at all. Knowing the benefits of the right solution matters just as much as knowing the pitfalls of the wrong one; you now have a solid grasp on both.

It wasn’t Mike that caused a pileup with his addition of a method, it was starting with the wrong pattern. Bad decisions compound. So set the standard: make a consumer-defined interface today so you aren’t fighting someone else’s 3000-line producer-defined interface six months from now.

Next time you’re adding a database method to a shared repository, setting up a new feature, or want to refactor that 500-line Jenga block, pause for a second. Ask yourself: “does every function in my codebase need LaunchUserIntoSpaceQuery(), or just the one I’m working with?”

I’m Ivan, follow me for more content like this! :)

Stop Writing try/except Hell: Clean Database Transactions with SQLAlchemy with the Unit Of Work

Ivan Korostenskij — Thu, 11 Dec 2025 00:37:49 +0000

Every codebase has that service function - the one drowning in try/except blocks, manual rollbacks, and session management that makes you question your career choices:

async def create_order(user_id: int, items_in_basket: list[dict]) -> Order:
    session = get_session()
    try:
        user: User = await session.get(User, user_id)
        if not user:
            await session.rollback()
            raise ValueError("User not found")

        order: Order = Order(user_id=user_id)
        session.add(order)

        for item in items_in_basket:
            line: OrderLineItem = OrderLineItem(**item, order=order)
            session.add(line)

        await session.commit()
        return order
    except DuplicateError:
        await session.rollback()
        raise
    except Exception:
            await session.rollback()
            raise
    finally:
        await session.close()

You've got rollback calls scattered everywhere, a finally block you'll forget to add next time, and business logic buried under boilerplate. It's fragile, verbose, and one missed rollback() away from a corrupted database state.

Let me show you a pattern that reduces this to:

async def create_order(user_id: int, items_in_basket: list[dict], uow: UnitOfWork) -> Order:
    async with uow:
        user = await uow.session.get(User, user_id)
        order = Order(user_id=user_id)
        uow.session.add(order)

        for item in items_in_basket:
            uow.session.add(OrderLine(**item, order=order))

        await uow.commit()
        return order

No manual rollback. No finally. No way to forget cleanup. Let's build it.

What is a Unit of Work?

Martin Fowler defined it best:

"A Unit of Work keeps track of everything you do during a business transaction that can affect the database. When you're done, it figures out everything that needs to be done to alter the database as a result of your work."

Think of it like an online shopping cart. You add items, remove items, change quantities - but nothing hits your credit card until you click "Purchase." If you abandon the cart, nothing happens. That's Unit of Work for databases:

Track changes in memory
Commit everything, all at once at the end, or
Rollback everything if something fails

How would you like it if you ordered online groceries to make a pizza, only for your delivery person to come back with just mozzarella, oregano, and raw flour?

Exactly. Your database work needs to be atomic (all-or-nothing). This pattern further provides automatic cleanup and separation between your business logic and persistence mechanics.

The Implementation

Here's the complete, copy-pastable solution:

from typing import Self
from types import TracebackType
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker

class UnitOfWork:
    """Async context manager for database transactions.

    Commits on success, rolls back on exception, always cleans up.
    """

    def __init__(self, session_factory: async_sessionmaker[AsyncSession]) -> None:
        self._session_factory = session_factory

    async def __aenter__(self) -> Self:
        self.session = self._session_factory()
        return self

    async def __aexit__(
        self,
        exc_type: type[BaseException] | None,
        exc_val: BaseException | None,
        exc_tb: TracebackType | None,
    ) -> None:
        if exc_type is not None:
            await self.rollback()
        await self.session.close()

    async def commit(self) -> None:
        await self.session.commit()

    async def rollback(self) -> None:
        await self.session.rollback()

Here’s how it works:

__aenter__ creates a fresh session when you enter the async with block. Each transaction gets its own isolated session.

async with UnitOfWork() as uow:
        ...
        # the `async with` calls __aenter__, and 
        #  returns our uow object that contains out db session

__aexit__ handles cleanup automatically. The magic is in exc_type - Python passes exception info if something went wrong inside the block. If there's an exception, we rollback. Either way, we close the session.

async with UnitOfWork() as uow:
        users: list[User] = uow.find_all_banned_users()

print(users) # When we "leave" that `async with` block, the database 
                         #  session is closed!

Explicit commit() keeps you in control. You decide when changes persist. No surprises.

"Why do I have to call commit() manually?"

You might be tempted to put self.commit() inside the __aexit__ method so it saves automatically if no errors occur; this is common, but misguided practice.

Explicit commits are intentional:

They prevent accidental writes if you return early from a function
They make it obvious to anyone reading your code exactly when the data is saved
You don’t waste resources automatically committing (or worse, accidentally commit something) on a simple select statement where you only read

As always: explicit is better than implicit

https://peps.python.org/pep-0020/

Using It

First, set up your session factory (do this once at app startup):

from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncEngine

engine: AsyncEngine = create_async_engine(
        "postgresql+asyncpg://user:pass@localhost/db"
        )

SessionFactory = async_sessionmaker(engine, expire_on_commit=False)

Then inject it into your UnitOfWork:

async def main() -> None:
    uow: UnitOfWork = UnitOfWork(session_factory=SessionFactory)

    async with uow:
        uow.session.add(User(name="Alice", email="alice@example.com"))
        await uow.commit()

        print(f"Created user with ID: {user.id}")

Here's what happens in different scenarios:

Happy path - code runs, you call commit(), changes persist:

async with uow:
    uow.session.add(User(name="Bob"))
    await uow.commit()  *# Saved to db*

Exception before commit - automatic rollback, nothing persists:

async with uow:
    uow.session.add(User(name="Charlie"))
    raise ValueError("Something went wrong")
    await uow.commit()  *# Never reached# __aexit__ catches the exception, 
                                            #  calls rollback(), Charlie never existed in the database*

Forgot to commit - nothing persists (safe default):

async with uow:
    uow.session.add(User(name="Dave"))
    *# Oops, forgot commit(), session closes, uncommitted changes discarded*

Real-World Example: Order Processing

Let's look at a more complete, realistic scenario - creating an order with multiple line items where any failure should cancel the entire operation:

from dataclasses import dataclass

@dataclass
class OrderRequest:
    user_id: int
    items: list[dict]

async def process_order(request: OrderRequest, uow: UnitOfWork) -> Order:
    async with uow:
        *# Validate user exists*
        user = await uow.session.get(User, request.user_id)
        if not user:
            raise ValueError(f"User {request.user_id} not found")

        *# Create order*
        order = Order(user_id=request.user_id, status="pending")
        uow.session.add(order)

        total = 0
        for item in request.items:
            product = await uow.session.get(Product, item["product_id"])
            if not product:
                raise ValueError(f"Product {item['product_id']} not found")

            if product.stock < item["quantity"]:
                raise ValueError(f"Insufficient stock for {product.name}")

            line = OrderLine(
                order=order,
                product_id=product.id,
                quantity=item["quantity"],
                price=product.price
            )
            uow.session.add(line)

            product.stock -= item["quantity"]
            total += product.price * item["quantity"]

        order.total = total
        await uow.commit()
        return order

If the user doesn't exist? Rollback. Product not found? Rollback. Insufficient stock? Rollback. The order, line items, and stock changes all succeed together or fail together.

With proper atomicity (the A in ACID), we always roll back grouped transactions if a single one fails to prevent corrupted or incomplete data.

Important Nuance

"Fancy algorithms are slow when n is small, and n is usually small." — Rob Pike

Not every database operation needs a Unit of Work wrapper. If you're writing a simple script that runs once, or a function that does a single INSERT, the overhead isn't worth it.

Unit of Work shines when:

Multiple related operations must succeed or fail together
You're building a layered architecture with services and repositories
You need testable transaction boundaries
Different parts of your code need to participate in the same transaction

For a quick one-off query? Just use the session directly.

Conclusion

That's it. You now have a pattern that:

Guarantees cleanup with context managers
Rolls back automatically on exceptions
Makes transaction boundaries explicit and testable
Separates business logic from persistence boilerplate

The Unit of Work pattern isn't about adding complexity - it's about removing the mental overhead of "did I remember to rollback?" from every line of ORM code you write.

Go find a service function with scattered try/except/finally blocks. Refactor it. Watch the noise disappear.

Questions about Unit of Work or SQLAlchemy patterns? Drop a comment below. :)

5 Essential Go Utility Functions: Filter, Map, and More (With Examples)

Ivan Korostenskij — Thu, 04 Dec 2025 03:24:57 +0000

Did you know Go only has 25 keywords (e.g., if, for, return)? Python has 38, and Java has 67. This represents the rest of Go: a minimal, approachable language that allows the quick prototyping of Python at the speed of C++. This simplicity comes at a cost: it’s common to feel like common core functionality is missing, especially if you’re coming from other languages.

Some commonly used utilities not included in Go

Map: Take a list of items, apply something to each one
Filter: Take a list of items, filter them all by a condition
Unique: Given a list, return unique items

The Go team has long argued most requests were too “trivial" - and that users should use an if statement. Only in late 2023 did they add the min() and max() functions.

However, we learn early on in programming - explicit, declarative, human-readable code is the way to go, not bunches of nested for and if blocks.

In this article, we’ll create and understand how to make 5 utility functions that’ll transform your code into reusable, clean, declarative expressions. You’ll be able to start using these immediately (I highly encourage it) - fitting into any workflow, all under 20 lines of code.

The preview:

Ternary
Filter
Map
Unique
Must

Perform a ternary operation

Run code my for this section right here: https://go.dev/play/p/k9WTQFGUD96

Coming from most other programming languages, you're probably used to inline conditionals:

const temperature = 15;
const targetTemp = 20;

const roomStatus = temperature < targetTemp ? "Heating" : "Stable";

Or Python’s version:

temperature: int = 15
target_temp: int = 20

room_status: str = "Heating" if temperature < target_temp else "Stable"

In Go, there's no built-in ternary operator - which feels limiting at first, but we can build one that's actually more flexible:

func Ternary[T any](condition bool, trueVal T, falseVal T) T {
    if condition {
        return trueVal
    }
    return falseVal
}

We’ll break down the[T any] syntax in a moment.

Let's see this in action. Instead of writing:

shipName := ""
defaultName := "The Flying Dutchman"
finalName := ""

if shipName != "" {
    finalName = shipName
} else {
    finalName = defaultName
}

We can express the same logic in a single line:

shipName := "" 
defaultName := "The Flying Dutchman"

finalName := Ternary(shipName != "", shipName, defaultName)

The result is more readable, reusable, and declarative: the code describes what we want, not the mechanics of how to get it.

Imperative vs Declarative

Imperative code describes how to do something step by step. Declarative code defines what outcome you want, hiding the implementation details behind a clear interface. Ternary(condition, a, b) immediately communicates intent, while the six-line if/else block makes you parse the logic yourself. Design your high-level code for humans, not compilers.

A quick note on generics (T any)

func Ternary[T any](condition bool, trueVal T, falseVal T) T {

The [T any] part is a type parameter, it tells Go that we don’t know what type trueVal and falseVal are going to be (e.g., string, int, bool ?), but we promise to fill those in when we call the function.

Without generics, we’d need separate functions for every type: TernaryString(), TernaryInt, TernaryBool...

With generics, one function handles them all, while ensuring type safety: Because type T any is the type of both trueVal and falseVal, our function will yell at us if we try to make them different:

name := Ternary(userIsLoggedIn, user.UserName, "Guest")         // T is string
count := Ternary(hasItems, itemCount, 0)                        // T is int
config := Ternary(isProd, prodConfig, devConfig)                // T is Config (object)

// Illegal (types of input variables don't match)
colorScheme: Ternary(isDarkMode, "background-black", false)

Go infers T from the arguments you pass - it’s smart enough to understand T is a string, int, whatever… without you needing to explicitly tell it. So, our utility function now adapts to whatever we throw at it.

Filter

Arguably the most useful utility here, Filter takes a list and a function, returning a list of elements that pass your filter’s test.

Here’s how we’ll do it:

func Filter[T any](collection []T, predicate func(item T, index int) bool) []T {
    result := make([]T, 0, len(collection))
    for i, item := range collection {
        if predicate(item, i) {
            result = append(result, item)
        }
    }
    return result
}

In practice:

activeUsers := Filter(users, func(u User, _ int) bool {
    return u.Status == "active"
})

Three things make this implementation both efficient and reusable:

Using generics keep this flexible

func Filter[T any](collection []T, predicate func(item T, index int) bool) []T

The type parameter T again means this works with strings, integers, maps, and even custom objects - while still retaining type-safety based on our constraint:

a list of T must be provided
a type of T must be the input to our filtering function we pass (predicate)
a list of T must come out

Pre-allocating our array makes this fast and memory efficient

Starting with a simply array and adding an item to it:

fruitList := []str{"apple", "tomato?", "kiwi"}

fruitList = append(fruitList , "banana")

Every time we exceed the list’s current capacity (3 things), Go has to:

Make an entirely new array (usually doubles in size as a guess!)
Copy everything from the previous array over to the new one, ordered
Return you the new list

So, we’ll tell Go “Create a list of type T , which is currently empty, but save space for up to len(collection) items - that’s the max.”


result := make([]T, 0, len(collection))

The index parameter allows us flexibility of using the index of the list elements to filter

While you can ignore the index with _, having it available lets you handle position-based filtering when needed:

// Keep only the first 5 items
first5 := Filter(items, func(item Item, i int) bool {
    return i < 5
})

// Keep every other item
evens := Filter(items, func(item Item, i int) bool {
    return i%2 == 0
})

// Combine position and value logic
validRecords := Filter(records, func(r Record, i int) bool {
    if i < 10 {
        return r.Priority == "high"
    }
    return r.Status == "active"
})

Or just ignore it:

activeUsers := Filter(users, func(u User, _ int) bool {
    return u.Status == "active"
})

Tip: Extract your filtering functions

“Any fool can write code that a computer can understand. Good programmers write code that humans can understand.”

— Martin Fowler (from Refactoring: Improving the Design of Existing Code)

Inline lambdas, although convenient code golf, are hard to read and impossible to test. Instead of this:


validProducts := Filter(products, func(p Product, _ int) bool {
    return slices.Contains(allowedBrands, p.Brand) &&
           p.Status != "discontinued"
}

Extract the logic into a named function (add early returns for extra readability!):

func isValidProduct(p Product, _ int) bool {
    if !slices.Contains(allowedBrands, p.Brand) {
        return false 
    }
    if p.Status == "discontinued" {
        return false
    }
    return true
}

validProducts := Filter(products, isValidProduct)

Now it's testable in isolation and readable at the call site.

Map

Where Filter shortens a list, Map transforms it, element by element - often into different types.

// Map transforms elements from type T to type R.
func Map[T, R any](collection []T, transform func(item T, index int) R) []R {
    result := make([]R, len(collection))
    for i, item := range collection {
        result[i] = transform(item, i)
    }
    return result
}

We use it like this:

// Extract: Take a list of users and extract their IDs into a list
userIDs := Map(users, func(u User, _ int) int64 { return u.ID })

// Transform: Take a list of Orders and return them along with their total cost
responses := Map(orders, func(o Order, _ int) OrderResponse {
    return OrderResponse{ID: o.ID, Total: o.CalculateTotal()}
})

The same efficiency gains are made here as from Filter :

Pre-allocating the exact result size (make([]R, len(collection))) eliminates append overhead entirely since we know the output length
The dual type parameters [T, R any] allow transforming between any types
1. Users to IDs
2. Orders to API responses
3. Raw data to nicely-structured DTOs

Extract your transform functions too

func toUserID(u User, _ int) int64 { 
    return u.ID 
}

userIDs := Map(users, toUserID)

func toOrderResponse(o Order, _ int) OrderResponse {
    return OrderResponse{
        ID:    o.ID, 
        Total: o.CalculateTotal(),
    }
}

responses := Map(orders, toOrderResponse)

Explicit is better than implicit; simple is better than complex; and declarative is better than imperative:

Before: responses := Map(orders, func(...) { ... something something complex logic ... })
- What is this doing?
After: responses := Map(orders, toOrderResponse)
- Ah ok, we’re transforming orders into API response objects. Intent is immediate.

No matter what language you’re working with, take this to heart:

https://peps.python.org/pep-0020/

Unique: Filtering out unique items efficiently with a Go set implementation

Ever noticed how there isn’t a built-in Setdata structure?

A mutable, unique array with an O(1) lookup, add, remove, and contains operations

We’ll build one with a map of struct{} values. Let’s say your key, in this map, is a string (16 bytes), you might think that to make this a map we have to store something with the value; we don't. Instead, we can store an empty struct (a struct{}{} - meaning a blank, uninitialized variable), which takes 0 bytes! So the map just tracks which keys exist, not anything else about them.

Go's compiler optimizes away empty structs entirely - they have no fields, so there's literally nothing to store in memory

It’s like going to a librarian, asking them if they have a copy of Moby Dick, getting an answer within 100 milliseconds, but getting no further information where it is. For a set - where all we need to do is check inclusion, add, and remove unique keys, this is all we need.

func Uniq[T comparable](collection []T) []T {
    seen := make(map[T]struct{}, len(collection))
    result := make([]T, 0, len(collection))

    for _, item := range collection {
        if _, exists := seen[item]; !exists {
            seen[item] = struct{}{}
            result = append(result, item)
        }
    }
    return result
}

We use it like this:

uniqueTags := Uniq([]string{"go", "rust", "go", "python", "rust"})

// Result: ["go", "rust", "python"]

Notice our function takes a different type of generic - T comparable instead of T any. This is a type constraint: meaning that the items we store (the keys) have to allow == and != operations, which is crucial for our set implementation.

Practically, this means that this will work with strings, numbers, booleans, and objects (structs) with comparable fields, but not lists or other maps (which don’t support == comparison in Go).

Internally, this implementation works by: for each item, checking if it exists in our map seen (pre-allocated!). If it’s found, skip it, otherwise add it to 1) our map seen and 2) our output list we’ll return result

if _, exists := seen[item]; !exists {
    seen[item] = struct{}{}
    result = append(result, item)
}

Must: panic if something goes catastrophically wrong

This is something I use frequently to cut down on initialization boilerplate.

Most applications will have startup code that looks like this:

cfg, err := loadConfig(cfgPath)

if err != nil {
    return fmt.Errorf("failed to load config: %w", err)
}

Logger, err = setupLogger(cfg.LogLevel)
if err != nil {
    return fmt.Errorf("failed to setup logger: %w", err)
}

DB, err = sql.Open(cfg.DBDriver, cfg.DSN)
if err != nil {
    return fmt.Errorf("failed to open DB connection: %w", err)
}

// Make sure the server is responsive
response, err = DB.Ping()
if err != nil {
    DB.Close()
    return fmt.Errorf("failed to ping DB: %w", err)
}

This error handling is verbose and arguably pointless. If any of these fail, your app can’t run anyway immediately fail. Why bubble errors up through initialization when you could just fail immediately?

Let’s clean this up to reduce boilerplate and increase readability with this utility function:

func Must[T any](v T, err error) T {
    if err != nil {
        panic(err)
    }
    return v
}

Now initialization becomes:

AppConfig = Must(loadConfig(cfgPath))

Logger = Must(setupLogger(AppConfig.LogLevel))

DB = Must(sql.Open(AppConfig.DBDriver, AppConfig.DSN))
Must(DB.Ping())

Clean, readable, and explicit about intent: these operations must succeed, or we fail fast.

Warning: Use Must strictly for initialization of critical services your app cannot function without. Do not use it in regular application logic - that's where proper error handling belongs.

Learn here, but use `lo` - a Go library dedicated to production-ready utility functions

The point of this article was to walk through idiomatic, well-engineered implementations of common Go utility functions. Understanding how they work will make you a better Go developer - but that doesn’t mean you should reinvent the wheel every project.

https://github.com/samber/lo/

Lo, by samber, is a battle-tested Go utility library that provides a wide array of smartly-optimized, robust helper functions like the ones we made today, plus about 100 more:

Filter
Map
Reduce
ForEach
Flatten
First / FirstOrEmpty
Uniq / UniqBy
GroupBy / GroupByMap
Chunk / PartitionBy
…

Use this article to understand the patterns and customize to your use case. If you find yourself needing more, use lo in production.

Putting it together

Let's use these utilities compose in a practical example: imagine we're processing hashtags from a social media post - we need to clean them up, validate them, and deduplicate them:

func processHashtags(raw []string) []string {
    normalizedHashtag := Map(raw, normalizeHashtag)
    validHashtags := Filter(normalizedHashtag, isValidHashtag)
    return Uniq(validHashtags)
}

func normalizeHashtag(tag string, _ int) string {
    tag = strings.TrimSpace(tag)
    tag = strings.ToLower(tag)
    tag = strings.TrimPrefix(tag, "#")
    return tag
}

func isValidHashtag(tag string, _ int) bool {
    if len(tag) == 0 || len(tag) > 30 {
        return false
    }
    if slices.Contains(bannedWords, tag) {
        return false
    }
    return true
}

Usage:

raw := []string{"  #GoLang ", "#RUST", "#go", "#ThisTagIsWayTooLongToBeUseful", "#spam"}
bannedWords := []string{"spam", "promo"}

result := processHashtags(raw)
// Result: ["golang", "rust", "go"]

Notice how readable the processHashtags function at first glance:

We normalize hashtags - which includes trimming white space, lowercasing, and removing the “#”
We remove invalid hashtags - those who are either too short or too long; or contain a banned word
We de-duplicate them

Boom, this is the payoff of building smart utilities and writing declarative code - our business logic reads like a description of what it does.

Wrapping up

That’s it!

You now have a toolkit of patterns that will serve you well beyond these specific functions:

Pre-allocation to avoid repeated memory allocations
Empty structs (struct{}{}) for zero-cost set implementations
Generics to write utilities once and use them everywhere
Extracted filter and transform functions to keep code testable and intent clear

More importantly, you've seen how to write Go that's both performant and readable - because the best code is code your teammates (and future you) can quickly understand and iterate on.

“Code is read much more often than it is written.* We are constantly reading code for comprehension, to fix bugs, or to add new features. Making sure your code is as readable as possible will pay dividends for the entire lifecycle of the program.”*

— Guido van Rossum (made Python)

If this helped you level up your Go, I'd love to hear about it! Drop a comment about a topic in Go or Python you’re interested in me covering. And if you want more practical Go content like this, give me a follow - I'm just getting started :)

The Shopify products.json Trick: Scrape Any Store 25x Faster with Python

Ivan Korostenskij — Thu, 27 Nov 2025 06:14:44 +0000

If you’ve ever done any web scraping, you know how annoying it is to parse HTML: finding parents of elements, navigating through a web of xml tags, hoping CSS selectors don’t change over time and break everything.

However, on Shopify - one of the largest and most useful repositories of product data - contains a trick that allows us to bypass this process entirely, leaving us with beautiful JSON right out the gate.

I recently built a fragrance drop tracker that scrapes dozens of Shopify
stores in real-time. Let’s walk through how and why this trick made that trivial.

The secret is `/products.json`

Every Shopify store exposes a secret endpoint: /products.json.

For perspective, I invite you to click on some links (promise they’re safe)

This is a website that sells niche perfumes in the spirit of animals (Cow, Penguin, T-rex…)

This first link is their direct home page

https://www.zoologistperfumes.com/

Now check this out:

https://www.zoologistperfumes.com/products.json?limit=10&page=1

{
  "products": [
    {
      "id": 1234567890,
      "title": "Bee (2023)",
      "handle": "bee-2023",
      "vendor": "Zoologist",
      "product_type": "Extrait de Parfum",
      "tags": ["honey", "beeswax", "floral"],
      "variants": [
        {
          "id": 9876543210,
          "title": "60ml",
          "price": "195.00",
          "available": true,
          "sku": "ZOO-BEE-60"
        }
      ],
      "images": [
        {
          "src": "https://cdn.shopify.com/...",
          "width": 1000,
          "height": 1000
        }
      ]
    }
  ]
  ... bunch of more stuff here ...
}

That’s money. Let’s walk through how we can fetch every product off this site in under a second with Python.

The code below is copy-pastable, and I’d highly encourage you to follow along as once you understand this, web scraping becomes a whole lot easier.

import asyncio

import httpx
from pydantic import BaseModel

# Define a few models so the nicely-formed JSON flows
#    smoothly into Python objects instead of opaque dictionaries
class ShopifyVariant(BaseModel):
    id: int
    title: str
    price: str
    available: bool
    sku: str

class ShopifyProduct(BaseModel):
    id: int
    title: str
    vendor: str
    tags: list[str]
    variants: list[ShopifyVariant]

class ShopifyResponse(BaseModel):
    products: list[ShopifyProduct]

async def fetch_product(client: httpx.AsyncClient, url: str) -> ShopifyResponse:
    response: httpx.Response = await client.get(url)

    return ShopifyResponse(**response.json())

async def scrape_shopify_products(base_url: str, client: httpx.AsyncClient) -> list[ShopifyProduct]:
    all_products: list[ShopifyProduct] = []
    page_number: int = 1

    while True:
        products: ShopifyResponse = await fetch_product(
            client=client,
            url=f"{base_url}/products.json?page={page_number}"
        )

        if len(products.products) == 0:
            break

        all_products.extend(products.products)
        page_number += 1

    return all_products

async def main() -> None:
    async with httpx.AsyncClient() as client:
        products: list[ShopifyProduct] = await scrape_shopify_products(
            base_url="https://www.zoologistperfumes.com",
            client=client
        )

    print(f"Found {len(products)} products")
    # Output: Found 115 products

if __name__ == "__main__":
    asyncio.run(main())

Want a better understand of how to efficiently batch mass amounts of async API calls in Python? Check out this article:

What Modern Python Uses for Async API Calls: HTTPX & TaskGroups

Ivan Korostenskij
Ivan Korostenskij

Ivan Korostenskij

Follow

Nov 26 '25

What Modern Python Uses for Async API Calls: HTTPX & TaskGroups

#python #tutorial #learning #productivity

11 reactions
Comments Add Comment

7 min read

Pydantic SOLVES the issue of parsing JSON in these scenarios, use it

By setting up these Pydantic models, we give ourselves automatic serialization of the raw JSON payload, documentation of exactly how the payload looks, and type-safe data access in our editors.

Plus, when you want to transform this data, you’re working with structured objects instead of dict soup.

Here’s how to scrape 5 sites in the same time as 1

async def main() -> None:
    shopify_sites_list: list[str] = [
        "https://www.zoologistperfumes.com",
        "https://bruvi.com/",
        "https://flourist.com/en-us",
        "https://packagefreeshop.com/",
        "https://meowmeowtweet.com/",
    ]

    products: list[ShopifyProduct] = []

    async with httpx.AsyncClient() as client:
        try:
            async with asyncio.TaskGroup() as task_group:
                tasks: list[asyncio.Task[list[ShopifyProduct]]] = [
                    task_group.create_task(
                        scrape_shopify_products(
                            base_url=site,
                            client=client
                        )
                    )
                    for site in shopify_sites_list
                ]
            for task in tasks:
                products.extend(task.result())

        except* Exception as eg:
            for error in eg.exceptions:
                print(f"Got an error: {error}")

    print(f"Found {len(products)} products")
    # Output: Found 626 products

if __name__ == "__main__":
    asyncio.run(main())

We’re adding a task group that gathers and executes calls to each website asynchronously, so no single website with 1000 pages blocks any others.

Make scraping 25x faster with binary search

Right now, our sequential scraper has to check page 1, page 2, page 3... until it hits an empty page. That's slow. This story explains how we fix that:

A Cambridge computer science professor got his bike stolen on campus one weekend. He went to the police and was relieved to find that they had a camera posted in plain view of his missing bike.

However, the police argued that they didn't have time to sift through all the footage, so he was out of luck.

So, he tried arguing, telling them:

If they forwarded the footage to Sunday night and the bike hadn’t been stolen, that meant it was stolen Sunday (24 hour window)
Repeating this again, but at noon on Sunday: if bike was still there, it had to have been stolen between 12pm-12am that day (12 hour window)
Doing this 7 more more times, you narrow down a 5 minute window where the thief shows up

After they refused to listen, he angrily made the point that if they had to sift through camera footage since the cretaceous period, it would’ve taken less than an hour - following his approach - to find the exact second of theft.

Figure 1: Comparison of binary (top) vs sequential (bottom) search, both searching for the target number 37, with binary search drastically winning out. This target number represents the last page of products in the sites we're scraping.

Let’s apply this. Right now, our algorithm doesn’t know how many pages there are, so instead of unleashing a hoard of API calls to collect the data concurrently, we have to tip toe around the unknown final page.

If we first:

Find that last page through binary search
Cache it

We know the stopping point and send dozens of concurrent requests to grab the data, drastically speeding things up.

async def find_last_page(base_url: str, client: httpx.AsyncClient) -> int:
    low, high = 1, 1000
    last_valid_page = 1

    while low <= high:
        mid = (low + high) // 2
        url = f"{base_url}/products.json?page={mid}"

        response: httpx.Response = await client.get(url)
        data: ShopifyResponse = ShopifyResponse(**response.json())

        if data.products:
            last_valid_page = mid
            low = mid + 1  # Try higher
        else:
            high = mid - 1  # Try lower

    return last_valid_page

async def scrape_all_pages_fast(base_url: str, client: httpx.AsyncClient) -> list[ShopifyProduct]:
    last_page: int = await find_last_page(base_url, client)
    all_products: list[ShopifyProduct] = []

    async with asyncio.TaskGroup() as tg:
        tasks: list[asyncio.Task[ShopifyResponse]] = [
            tg.create_task(
                fetch_product(client, f"{base_url}/products.json?page={page}")
            )
            for page in range(1, last_page + 1)
        ]

    for task in tasks:
        all_products.extend(task.result().products)

    return all_products

Here are the numbers:

Sequential Approach: We scrape page by page until we find the last one

Binary Search Approach: We find the last page with binary search, then concurrently scrape every page

Math:

I tested 10 websites at once averaging 23 pages (600 products) each.

The first time, the sequential approach took an average 3.10s while the binary approach took almost double - 5.926s.

The following runs, once that last page was cached, the binary approach scraping went 5.94s -> 0.23s, a 25.8x increase!!

In smaller runs, the investment of binary search pays off after 2 runs, while on larger tasks it wipes the floor.

Stress test

I also found a site with the max amount of products (25000, 833 pages, "https://pelacase.ca/").

	Sequential	Binary
First Run Time	120.1s	11.56s
Second Run (cached)	101.4s	5.26s

Important Nuance

Custom Shopify sites sometimes seem to turn this off, so look before you scrape

https://cowboy.com/products.json

"Fancy algorithms are slow when n is small, and n is usually small."

Rob Pike

Don’t prematurely optimize what you don’t need to. A simple solution to the potential mass page problem, instead of concurrent pagination + binary search, is setting the item limit of the page you’re scraping to something ridiculous. Then you only have to scrape 1 page (given many other tradeoffs).

www.zoologistperfumes.com/products.json?limit=1000000&page=1

Binary search is just another tool. You won't always need it, but knowing it exists changes what problems you can now solve efficiently when the time comes.

Conclusion

That’s it. You know:

The backdoor products.json endpoint
How to scrape multiple sites concurrently (TaskGroups, async)
How to optimize pagination with binary search

More importantly, you learned that to optimize something, you often need to invest time illuminating it’s boundaries.

So pick a Shopify site, run the code, and watch it pull hundreds of products/second; add binary search and watch it get 25x faster.

Questions about web scraping or async Python? Drop a comment below.

Follow for more tutorials on building real production tools. :)

What Modern Python Uses for Async API Calls: HTTPX & TaskGroups

Ivan Korostenskij — Wed, 26 Nov 2025 02:27:15 +0000

Multiple API calls in Python are usually written in a way that makes them slow

You’ve written before:

import requests
from requests import Response

urls: list[str] = ["https://api.example.com/user/1", 
        "https://api.example.com/user/2",
        "https://api.example.com/user/3"]

for url in urls:
    response: Response = requests.get(url)
    print(response.json())

You run it. Request 1 goes out. Wait. Response comes back. Request 2 goes out. Wait…

Since these requests process one at a time, it's like sending texts to a friend, then staring at your phone - unblinking, refusing to eat, breathe, or move until they reply 'lol'.

Let me show you a better way. A way we can make these 100 requests take as long as your single slowest request.

Figure 1: Side-by-side comparison: Sequential requests (left) vs concurrent async requests (right). Sequential processes 10 API calls one-at-a-time taking 20 seconds total. Concurrent fires all 10 simultaneously, completing in 3 seconds - as fast as the slowest request.

00:00

Ivan Korostenskij

Async vs Sync Batch API Calls

#programming #beginners #python #tutorial

Comments

1 min read

Use asynchronous programming for API calls

Most time is spent waiting between calls. The solution is making them asynchronous - fire them off all at once and move on until responses come back.

API calls in Python are traditionally and ubiquitously done with the requests library.

import requests

def get_dad_joke() -> str:

        response = requests.get("https://icanhazdadjoke.com/")

        return response.text

for _ in range(3):
        joke: str = get_dad_joke()
        print(joke)

# Output:
    # Why do pirates not know the alphabet? They always get stuck at "C".
    # You can't run through a camp site. You can only ran, because it's past tents.
    # Now you can talk about Botox and nobody raises an eyebrow.

Unfortunately, the requests library doesn’t support async requests function calling.

The modern solution is a library called httpx

So pull up your IDE and let’s walk through how to easily start from scratch or migrate an existing project and tangibly see those performance gains.

From requests to httpx

For existing projects using requests wanting to migrate, the good news is that httpx is a near drop-in replacement.

Let’s look at an example.

def get_dad_joke_requests() -> str:

    response = requests.get("https://icanhazdadjoke.com/")

    return response.text

async def get_dad_joke_httpx() -> str:

    with httpx.AsyncClient() as client:

        response = await client.get("https://icanhazdadjoke.com/")

        return response.text

💡 Fun fact

The default timeout for requests is infinity - your request will wait forever. In httpx, it's 5 seconds.

httpx also supports HTTP/2 (multiple requests over one connection) and has strong type hints that catch bugs before runtime.

These improvements come from decades of production lessons learned.

The core difference is the introduction of the AsyncClient() . This is a session object: a dedicated state manager for the requests that are going to be made between your computer and the API server.

Then the async with block acts as an automatic door - it opens when you approach, closes when you leave, and you never have to think about it. In our case, it’s automatically handling session.open() and session.close() calls.

One step up from this is passing the http.AsyncClient as a dependency:

async def get_dad_joke_httpx(client: httpx.AsyncClient) -> str:

        response = await client.get("https://icanhazdadjoke.com/")

        return response.text

This way, whatever function is calling this instantiates that client, instead of us making it over and over again in the API call function itself.

The resources a function uses should be handled at the highest possible level. In this case, that’s the client that we make requests through.

Use asyncio.TaskGroups, not asyncio.gather to execute multiple API calls at once

We’re getting away from for loops. Think about a race. The starting official doesn’t go up to every individual racer and tell them to start - he uses a starting pistol, telling everyone at once.

This is the traditional way of batch-executing multiple async calls:

import asyncio

from httpx import Response
import httpx

async def get_dad_joke_httpx(client: httpx.AsyncClient) -> str:

    response: Response = await client.get("https://icanhazdadjoke.com/")
        response.raise_for_status()
    return response.text

async def main() -> None:
    dad_jokes_amount: int = 10

    async with httpx.AsyncClient() as client:
        results: list[str | Exception] = await asyncio.gather(
            *[get_dad_joke_httpx(client=client) for _ in range(dad_jokes_amount)],
            return_exceptions=True
        )

    # Since `results` is not a list of either errors or dad jokes,
    #   we need to manually check and discern
    for result in results:
        if isinstance(result, Exception):
            print(f"Got an error: {result}")
        else:
            print(f"Successfully got dad joke: {result}")

asyncio.run(main())

The isinstance(result, Exception) is a major code smell - we shouldn’t have to check/guess what our return types are. This is exacerbated if we have custom exceptions:

if isinstance(result, RateLimitError):
        print("we got rate limited, slowing down...")
        ...
elif isinstance(result, APIDownError):
        print("API is down, retrying automatically in 10 minutes")
        ...
elif isinstance(result, Exception):
        print(f"Got an error: {result}")

Since Python 3.11 we can use two features together that handle this cleanly for us.

import asyncio

from httpx import Response
import httpx

async def get_dad_joke_httpx(client: httpx.AsyncClient) -> str:

    response: Response = await client.get("https://icanhazdadjoke.com/")

    response.raise_for_status()
    return response.text

async def main() -> None:
    dad_jokes_amount: int = 10

    async with httpx.AsyncClient() as client:
        try:
            async with asyncio.TaskGroup() as tg:
                tasks: list[asyncio.Task[str]] = [
                    tg.create_task(get_dad_joke_httpx(client=client))
                    for _ in range(dad_jokes_amount)
                ]

            # If we get here, we know NO errors have occurred
            for task in tasks:
                print(f"Successfully got dad joke: {task.result()}")

        # We hit this if ANY errors occurred - the whole batch fails as a group of errors
        except* Exception as eg:
            for error in eg.exceptions:
                print(f"Got an error: {error}")

asyncio.run(main())

Those two features are TaskGroups and Exception Groups

Here, TaskGroup lets us treat our batched API calls as one running process. If anything fails, the entire batch fails - we catch an “Exception group” of those errors and can loop through them akin to successful results.

Exception groups are defined in the code above as so, using a * (asterisk) next to the except keyword to catch that group of errors.

Why do we fail the group if just a single request failed?

It seems counter-intuitive, but we do it to “fail fast”.

Failing fast ensures your application state is binary: it either worked perfectly, or it didn't happen at all.

Here are two common scenarios:

If request 1 fails with a 401 unauthorized, the next 99 will too. This batch SHOULD fail as to not hammer the server. We use this first failure as a canary in a coalmine.
Most often, we’re working with operations that to be atomic. In this case, partial failures lead to corrupted data that requires manual intervention.
1. E.g., New employee onboarding:
  1. Create their email (fails)
  2. Create their Teams account (succeeds)
  3. Add them to payroll (succeeds)
2. In the case above, now we have an employee that is getting paid but isn’t able to log in. It’s much better to fail the whole batch and retry with a clean slate than have to debug a half-onboarded user

Important nuance - we can’t let 100s of API calls free into the world at once

While we could now make 1000 API calls instantly, we’re going to come into a big blocker of batch operations on public-facing APIs: rate limiting.

This is the problem:

async def main():
        # This will make 1000 requests at once
    tasks = [get_joke(client) for _ in range(1000)]
    await asyncio.gather(*tasks)

Best case scenario you get rate limited after the first 100 and your requests fail with 429 rate limited, worst case you get banned because your batch looks like a DDoS attack.

These limits (e.g., “20 requests a second”, “100 max concurrent connections”) prevent server overload and malicious attacks exploiting resource limits.

The way we get around this is with a Semaphore: a queue that limits the amount of active processes at a time.

import asyncio

from httpx import Response
import httpx

async def get_dad_joke_httpx(client: httpx.AsyncClient, semaphore: asyncio.Semaphore) -> str:

    async with semaphore:
        response: Response = await client.get("https://icanhazdadjoke.com/",
                                              headers={
                                                  "Accept": "text/plain"
                                              })

        response.raise_for_status()
        return response.text

async def main() -> None:
    dad_jokes_amount: int = 10
    sem = asyncio.Semaphore(5)

    async with httpx.AsyncClient() as client:
        try:
            async with asyncio.TaskGroup() as tg:
                tasks: list[asyncio.Task[str]] = [
                    tg.create_task(get_dad_joke_httpx(client=client,
                                                      semaphore=sem))
                    for _ in range(dad_jokes_amount)
                ]

            for task in tasks:
                print(f"Successfully got dad joke: {task.result()}")

        except* Exception as eg:
            for error in eg.exceptions:
                print(f"Got an error: {error}")

asyncio.run(main())

Now our code will be as fast as possible, just within the limits of the server we’re communicating with.

Conclusion

That's it. This is a common and extremely effective use of asynchronous programming. All that's left is to apply it. Even outside of Python, you're walking out of this with a seriously generalizable skill and I highly encourage you to look for opportunities to create and/or refactor real code to set this in.

Common opportunities I see to refactor:

Looping over API calls (usually batching calls)
An API relying on retries or a time.sleep() to avoid rate limits - use a semaphore!
Blocking, long, sync database queries - replace with async ones

Overall, start noticing blocking operations in your code - you'll often find that other things can be done while they're running.

One last thing:

Although I personally encourage asynchronous programming from the start of all new projects - you don't always need it. If you're making 3 API calls in a script you'll run once, requests is fine. Don't optimize for problems you don't have.

But when you do have that problem - watching requests crawl one by one, or notice yourself sitting infront of a slow progress bar a bit too often - you know what to do.

Now go find a loop of API calls in your codebase. Convert it. Time it before and after, brag a little, maybe share my article :)...

Thanks for reading!

Questions about async patterns or other Python stuff? Drop a comment. Follow for more tutorials on writing better Python.

Python Registry Pattern: A Clean Alternative to Factory Classes

Ivan Korostenskij — Sat, 22 Nov 2025 03:40:51 +0000

Every codebase has that one factory/builder nobody wants to touch. It starts out innocent, but two years and ten features later, it's mutated into a 500-line if-else monstrosity:

class FishFactory:

    def create_fish(self, fish_id: int, water_type: str, depth: int, is_holiday: bool) -> Fish:
        if fish_id == 1:
            return ClownFish()

        elif fish_id == 28:
            # TODO: have Brian fix this later - 06/12/2016
            if water_type == "SALT" or water_type == "SALTY":
                return BlueTang()
            elif water_type == "FRESH":
                return RoboticBlueTang()

        elif fish_id == 3:
            if depth > 100:
                return DeepSeaGramma()
            elif is_holiday:
                return RoyalChristmasGramma()

        elif fish_id == 4:
            # URGENT: Holiday hack for the 'Santa Goby' skin. Remove after Xmas 2023!!
            if is_holiday:
                return SantaHatGoby()
            return YellowGoby()

        elif fish_id == 6:
            # NOTE: ID 6 is now deprecated (the pufferfish incident), but we kept the code just in case
            return LongfinBlackMolly_DEPRECATED()

        elif fish_id == 7:
            # ... (200 more lines of this)
            return Cod()

        # Default fallback that shouldn't exist but does
        return GenericFish()


factory = FishFactory()
my_fish = factory.create_fish(fish_id=28, water_type="SALT", depth=50, is_holiday=False)

These factory classes are typically 1) critical to core logic and 2) frequently enhanced, which means they must be readable and maintainable. But they become the opposite - a dumping ground for every new edge case and feature flag.

But then how do we fix this? Some might try splitting it into multiple factories, others might reach for a strategy pattern. Both can work, but they often just move the complexity around rather than solving the root problem.

In real-world software engineering, the best answer is often the Registry Pattern.

A registry separates your lookup logic (the data structure that maps IDs to classes) from your routing logic (the business rules that decide which ID to use). This now:

Maintains the O(1) speed of a hash table
Enforces best practices for maintainable and clean code, preventing god objects by keeping concerns separated
Allows for runtime configuration through feature flags/plugins

Learning this pattern and the principles that guide it will increase your software’s maintainability, cohesion, and your co-workers impression of your technical skills :).

Common Traps: What You Should Avoid

Metaclass registry method

This registry implementation uses the init_subclass dunder method to automatically add classes that inherit from it into its store.

# --- registry.py ---
SHAPE_REGISTRY = {}

class BaseShape:
    def __init_subclass__(cls, key: str, **kwargs):
        super().__init_subclass__(**kwargs)
        SHAPE_REGISTRY[key] = cls

# --- shapes.py ---
from .registry import BaseShape

class Circle(BaseShape, key="circle"):
    ...

class Square(BaseShape, key="square"):
    ...

It’s a neat piece of code golf, but isn’t recommended because of how it “magically” registers classes, and is confusing to both debug and extend, hiding the logic of registration inside the parent class.

Decorator registry method

This method is a step up. It uses a custom decorator that explicitly “announces” registration of classes.

# --- registry.py ---
SHAPE_REGISTRY = {}

def register_shape(key: str):
    def decorator(cls):
        SHAPE_REGISTRY[key] = cls
        return cls
    return decorator

# --- shapes.py ---
from .registry import register_shape

@register_shape("circle")
class Circle:
    ...

@register_shape("square")
class Square:
    ...

# --- main.py ---
from .shapes import * # This import MUST happen (code smell already)
from .registry import SHAPE_REGISTRY

def create_shape(shape_name: str):
    return SHAPE_REGISTRY[shape_name]()

The issue is that decorators only run when the file containing them is *imported*. If you define Circle in shapes.py but never explicitly import that file in main.py, the decorator will not work - “magic” behavior you want to avoid. This forces you to import modules you aren't explicitly using just to trigger the side-effect of registration.

This method relies on:

Global state (debugging mess, hard to test).
You needing to import the file the original dictionary is in (causes annoying & unexpected bugs).
Doesn’t support registry editing at runtime or multiple registries.

At this point, I highly suggest opening up a Python file in your IDE and following along. You’ll learn the most by doing.

Tutorial: the right way

Let’s make a registry that can add and get key-value pairs. The most basic registry is a smart wrapper around a dictionary:

class Registry:
    def __init__(self):
        self._store: dict[str, str] = {}

    def register(self, key: str, value: str) -> None:

        self._store[key] = value

    def get(self, key: Key) -> Value | None:

        return self._store.get(key)

It’s important to have explicit error handling and we should probably support being able to remove a key:

def register(self, key: str, value: str) -> None:
    if key in self._store:
        raise ValueError(f"Key '{key}' is already registered.")
    self._store[key] = value

def get(self, key: Key) -> Value | None:
    # This means:
    #   "if variable `value` - which I'm assigning to be "self._store.get(key) -
    #.   exists, then return it, otherwise throw an error"
    if value := self._store.get(key):
      return value
    else:
      raise KeyError(f"Key '{key}' not found in registry.")

def unregister(self, key: Key) -> None
    if key not in self._store:
        raise KeyError(f"Key '{key}' not found in registry.")
    del self._store[key]

💡

Tip: If a function cannot do what its name says (e.g., get), it should throw an error, not return None.

Tip: Instead of adding a dedicated function to check if a key exists, implement the __contains__ dunder method

def __contains__(self, key: KeyType) -> bool:
            """Allows for `if Key in Registry` to work easily"""
        return key in self._store

Now we can do this:

my_registry: Registry = Registry()

registry.register("API_KEY", "abc12345")

if "API_KEY" in registry: 
    print("API_KEY is registered.")

# Output: "API_KEY is registered."

Make it Generic (Generics)

So far, our registry can only hold strings. If we needed to support an integer Key and an Object value, we’d need to re-create the entire class.

We can use Generics - placeholders for types we promise to fill in later - to give us first-class type safety

(Note: Syntax below requires Python 3.12+)

class Registry[Key, Value]:
    """Generic implementation of the Registry pattern"""
    def __init__(self):
        self._store: dict[Key, Value] = {}
    # ... (other methods now use Key and Value types)

Now we can create registries holding anything we want with ultimate type hinting safety

Integer / string registry

user_registry: Registry[int, str] = Registry()

user_registry.register(101, "Alice")
user_registry.register(205, "Bob")

print(f"Is key 101 registered? : {101 in user_registry}")
print(f"Value for 205: {user_registry.get(205)}")

String / Object (very common)

from dataclasses import dataclass

@dataclass
class ConfigItem:
    setting: str
    value: int

# Registry[Key: str, Value: ConfigItem]
config_registry: Registry[str, ConfigItem] = Registry()

# Register a single item
config_registry.register("TIMEOUT", ConfigItem("Network Timeout (s)", 30))

# Quick lookup and check
item: ConfigItem = config_registry.get("TIMEOUT")
print(f"Retrieved: {item}")
print(f"Is TIMEOUT registered? {"TIMEOUT" in config_registry}")

This is now flexible and gets you one step closer to the “LGTM” review from your senior dev.

Important: Features you should utilize

Now that you’ve learned the fundamentals, here are practical ways to customize and target your implementation of this in your project

1. Runtime Flexibility

Because our registry is just a living object, we gain runtime agility: we can add, remove, or swap handlers while the application is running.

Why and how?

Imagine you want to roll out a new feature flag. You want to switch between StripeV2 and StripeV3 as your payment handler without restarting your servers.

Tip: Do not rely on fully in-memory state as the source of truth. Changes are lost on reboot. Your code should query a database or feature flag system and use that signal to update your registry.

You should store the flag in your database/feature flag system

Id	FlagName	Value
1	stripe_v3	FALSE
2	stripe_v2	TRUE

Your code should query and use that signal to update your registry

# The Registry allows the application to react to the world
if config.is_feature_enabled("stripe_v3"):
    # We can swap implementation details on the fly!
    registry.register("payment", StripeV3)
else:
    registry.register("payment", StripeV2)

Dedicated matching logic

“What if I need more than a single string to match my values?“

”What if I need to match off 5 different parameters, including the > current weather and a random number between 1 and 20?

Instead of smashing it into a factory, use a Decision Engine.

A Registry should be a dumb lookup table. It shouldn't know about 'weather' or 'user mood'. Move complex logic into a dedicated decision engine that outputs a simple key.

Bad:

import random

def get_fishing_spot_from_complex_state(time: str, weather: str, mood: int) -> str:
        """Uncle dan's fishing spot algo
        """
    match (time, weather, mood):
        case ('morning', 'sunny', x) if x >= 15:
            return "Go to Great Heron Lake."

        case ('evening', _, 5):
            return "Go to Pond O' Wonders"

                # 20 more
                ... 

        case _:
            return "Stay home"

This acts as a hard-coded "address book." To add new entries, you have to modify this function (violating Open/Closed principle).

It also contains all the complex conditional logic to parse and interpret raw data (violates Single Responsibility principle)

Instead, split the decision maker (the matching logic) from the lookup (the registry):

from dataclasses import dataclass
from typing import Protocol, override

@dataclass
class SystemEvent:
    type: str
    amount: float = 0.0
    currency: str = "USD"

def decide_handler_key(event: SystemEvent) -> str:
    match event:
        case SystemEvent(type="payment", amount=amt) if amt > 1000:
            return "payment:high_value"
        case SystemEvent(type="payment"):
            return "payment:standard"
        case SystemEvent(type="new_user"):
            return "onboarding:new_user"
        case _:
            return "default"

class EventHandler(Protocol):
    def process(self, event: SystemEvent) -> None:
        ...

def handle_event(event: SystemEvent, registry: Registry[str, EventHandler]):

    routing_key: str = decide_handler_key(event)

    try:
        handler: EventHandler = registry.get(routing_key)

        handler.process(event)

    except KeyError:
        print(f"No handler registered for '{routing_key}'")

This separates the data (the mappings) from the logic (the application). Now we can test our business logic independently of our wiring.

Wiring It Up (plugins)

To register entries, use a dedicated registration function to be explicit about when and where registration happens.

# --- src/api/setup.py ---
def setup_app():
    # Explicit is better than implicit.
    # We can see exactly what is being loaded.
    handler_registry.register("payment", PaymentHandler)
    handler_registry.register("onboarding", OnboardingHandler)

Pro tip: Upgrade your registration process

You now easily support plugins to your system.

The registration logic should be modular, and live next to the code its registering.

This also makes it so when we need to add new entries, we’re not modifying main.py (following open/close)

# --- modules/shapes/registration.py ---

from .classes import Circle, Square, BaseShape
from ..registry import Registry

def register_shapes_module(registry: Registry[str, Type[BaseShape]]):
    """
    This is the "entry point" for the shapes module.
    It registers all classes this module provides.
    """
    print("Registering Shapes Module...")
    registry.register("circle", Circle)
    registry.register("square", Square)
    # When we add a Triangle, you ONLY edit this file

Main.py takes loader functions from specific modules and executes them to fill in the registries on startup

# --- main.py ---
from .registry import Registry
from .modules.shapes.registration import register_shapes_module
from typing import Callable

shape_registry = Registry()

# List of our "plugins" to load
MODULES_TO_LOAD: list[tuple[Callable, Registry]] = [
    (register_shapes_module, shape_registry),
    # 20 more here maybe
]

def setup_application() -> None:
    for register_func, registry_instance in MODULES_TO_LOAD:

        register_func(registry_instance)

setup_application()

Important Nuance

“In theory there is no difference between theory and practice, while in practice there is”

Not everything needs a generics-based, class-based, decoupled Registry system.

If you have a small, fixed list that doesn’t need to be changed at runtime, just use a dictionary.
If you’re writing a quick script or prototype, just use a if/elif function

Make it work, make it right, then make it fast.

Finish your feature, then let its pain points lead you to optimizations. But, knowing this is what helps you make those decisions when needed.

The Final Copy-Paste Solution

class Registry[Key, Value]:
    """Generic implementation of the Registry pattern using a hash map.
    """

    def __init__(self):
        """Initialize the registry with an empty internal store."""
        self._store: dict[Key, Value] = {}

    def register(self, key: Key, value: Value) -> None:
        """Register a key-value pair to the registry.

        This method checks for key collisions to ensure uniqueness.

        Args:
            key (Key): The unique identifier for the stored value.
            value (Value): The item (object, class, or callable) to store.

        Raises:
            ValueError: If the key is already registered.
        """
        if key in self._store:
            raise ValueError(f"Key '{key}' is already registered.")
        self._store[key] = value

    def get(self, key: Key) -> Value:
        """Retrieve a value by its key.

        Args:
            key (Key): The unique identifier to look up.

        Returns:
            Value: The stored value associated with the key.

        Raises: 
            KeyError: If the key is not found in the registry.
        """
        if value := self._store.get(key):
            return value
        else:
            raise KeyError(f"Key '{key}' not found in registry.")

    def unregister(self, key: Key) -> None:
        """Remove a key-value pair from the registry.

        Args:
            key (Key): The unique identifier of the item to remove.

        Raises:
            KeyError: If the key is not found in the registry.
        """
        if key not in self._store:
            raise KeyError(f"Key '{key}' not found in registry.")
        del self._store[key]

    def __contains__(self, key: Key) -> bool:
        """Checks if a key exists in the registry.

        This method enables the use of the 'in' operator (e.g., `if key in registry:`).

        Args:
            key (Key): The unique identifier to check.

        Returns:
            bool: True if the key is registered, False otherwise.
        """
        return key in self._store

    def list_keys(self) -> list[Key]:
        """Return a list of all keys currently registered.

        Returns:
            list[Key]: A list containing all registered keys.
        """
        return list(self._store.keys())

Usage example

@dataclass
class Fish:
    name: str

# Init
FishRegistry = Registry[int, Fish]()

# Register 
FishRegistry.register(1, Fish(name="Clownfish"))
FishRegistry.register(2, Fish(name="Blue Tang"))

# Lookup
try: 
    fish = FishRegistry.get(2)
    print(f"Found: {fish.name}")
except KeyError:
    print("Fish not found")

Forem: Ivan Korostenskij

Why Your Repository Pattern Creates Tech Debt (And How to Fix It in Python)

The Repository Pattern in Python

The (short) theory

Unit of Work

Stop Writing try/except Hell: Clean Database Transactions with SQLAlchemy with the Unit Of Work

A single decision determines whether this approach becomes pure tech debt or maintainable, readable code: where you put it.

The Bad Way

The good way

Nuance (with pushback)

A note on Go

The Repository Pattern Done Right: Consumer-Defined Interfaces in Go

Conclusion

The Repository Pattern Done Right: Consumer-Defined Interfaces in Go

The Repository Pattern

The (short) theory

Context: Further Reading

Stop Writing try/except Hell: Clean Database Transactions with SQLAlchemy with the Unit Of Work

A single decision decides whether this approach becomes pure tech debt or maintainable, readable code: where you put it.

The Bad Way

The good way

Nuance (with pushback)

Conclusion

Stop Writing try/except Hell: Clean Database Transactions with SQLAlchemy with the Unit Of Work

What is a Unit of Work?

The Implementation

Using It

Real-World Example: Order Processing

Important Nuance

Conclusion

5 Essential Go Utility Functions: Filter, Map, and More (With Examples)

Perform a ternary operation

Filter

Tip: Extract your filtering functions

Map

Extract your transform functions too

Unique: Filtering out unique items efficiently with a Go set implementation

Must: panic if something goes catastrophically wrong

Learn here, but use lo - a Go library dedicated to production-ready utility functions

Putting it together

Wrapping up

The Shopify products.json Trick: Scrape Any Store 25x Faster with Python

The secret is /products.json

What Modern Python Uses for Async API Calls: HTTPX & TaskGroups

Pydantic SOLVES the issue of parsing JSON in these scenarios, use it

Here’s how to scrape 5 sites in the same time as 1

Make scraping 25x faster with binary search

Important Nuance

Conclusion

What Modern Python Uses for Async API Calls: HTTPX & TaskGroups

Multiple API calls in Python are usually written in a way that makes them slow

Async vs Sync Batch API Calls

Use asynchronous programming for API calls

From requests to httpx

Use asyncio.TaskGroups, not asyncio.gather to execute multiple API calls at once

Important nuance - we can’t let 100s of API calls free into the world at once

Conclusion

Python Registry Pattern: A Clean Alternative to Factory Classes

Common Traps: What You Should Avoid

Metaclass registry method

Decorator registry method

Tutorial: the right way

Make it Generic (Generics)

Important: Features you should utilize

1. Runtime Flexibility

Dedicated matching logic

Wiring It Up (plugins)

Important Nuance

The Final Copy-Paste Solution

Usage example

Learn here, but use `lo` - a Go library dedicated to production-ready utility functions

The secret is `/products.json`