DEV Community: Kuba

Building Feature Flags: What I Learned Building My Own

Kuba — Sat, 27 Sep 2025 20:00:39 +0000

So I've been working on this project where the pace of development was much faster than our QA process. You know that feeling when you're shipping code daily, but your QA and client acceptance testing happens in batches? Yeah, that was my reality.

The problem hit me hard when I realized I was avoiding deploying changes I'd written a week or two ago. What happens when something breaks after tests, and I've already merged other stuff? Do I rollback everything? Cherry-pick commits? Deal with merge conflicts while the production is on fire?

Feature flags seemed like the answer to most of these headaches.

But here's the real kicker - I also wanted certain features to be configurable on the user level. Not just global on/off switches, but user-specific configuration. Like, I don't want to change the main behavior for all users, just for specific testers.

That's when I realized I needed two levels: global and user-specific control.

🎯 Why This Actually Mattered

Let me give you some real examples that'll make this concrete:

Payment Testing Nightmare 🤑

We wanted to test payments in production. The minimal hourly rate was set to $60. But I don't want to pay that much just to test payment flows (possibly several times). So I built user-level config to set min hourly rate. Now I can set it to $1 for testing. This is more like dynamic config than feature flags, but same mechanism.

New Credits System 💳

We introduced in-app credits, so payments weren't just Stripe anymore. Since I treat money as the most critical part of any app, I wanted an instant kill switch. Turn it off when things go wrong, turn it on when we're confident.

The Rollback Hell 🔥

You're always more prone to bugs when you're in a hurry. Release to prod, bug happens, you skipped some config, something works differently. You need to rollback, but you've already merged a bunch of changes. Merge conflicts during a production incident? That's a recipe for more mistakes.

With feature flags? Just flip the switch. No release process needed.

🙄 The Problem: Release Hell

The main problem of not having feature flags isn't that you're less flexible. The problem is that your release process is tightly coupled to your deployment process.

To change anything, you need to release a totally different version of your code. That means going through the entire testing process. Time is money, and this was burning both.

Here's my daily reality: I want to release code daily. I'm working on Feature A - let's say external calendar integration that impacts user availability. Since availability is core to the business process, it's critical.

There's no safe option to release each commit before completing this feature. So I stack all changes, finish everything, push the deploy button, and pray it doesn't blow up.

What if it does blow up?

Code changes → Commits → Testing → Release (DEV → TEST → PROD) → Back to square one

Here's the alternative I built:

Hide the impact behind feature flags from day one.

def is_available(self, user_id, datetime):
    if feature_flag_check('EXTERNAL_CALENDAR_INTEGRATION'):
        return check_availability_in_external_calendar(user_id, datetime)

    return check_basic_availability(user_id, datetime)

From the very first commit, I can push to PROD even if this function looks like:

def check_availability_in_external_calendar(user_id, datetime):
    return True  # TODO: implement actual logic

Because I'll just set EXTERNAL_CALENDAR_INTEGRATION to False.

Next step? Switch it to True for a specific user, test it, and if it blows up - flip it back to False instantly. No deployment, no CI, no QA process. Just faster and safer.

💡 My Solution: Keep It Simple, Keep It Fast

At this point, let's talk about how I built this. It's a simple mechanism to start with. You could use external solutions, but let's build a custom one that actually works.

What you need:

Storage for feature flags
Admin panel for CRUD operations
Tools to use them in your code

In my app, I'm using an in-memory event bus and the context pattern I described here. The key insight: Is a feature flag on/off? Depends on who's asking.

I store flags in a request-global Context variable. For each request, when building context, I fetch all feature flags enabled globally and those for the specific user.

class FeatureFlagsModel(BaseModel):
    __tablename__ = 'feature_flags'
    __table_args__ = (
        UniqueConstraint('key', 'user_id', name='uq_feature_flag_key_user'),
        Index('ix_feature_flag_user_id', 'user_id'),
        Index('ix_feature_flag_key', 'key'),
    )

    key: Mapped[ConfigKey] = mapped_column(String, nullable=False)
    enabled: Mapped[bool] = mapped_column(nullable=False, default=False)
    value: Mapped[Optional[str]] = mapped_column(nullable=True)

    user_id: Mapped[Optional[str]] = mapped_column(
        ForeignKey('[users.id](http://users.id)', ondelete='CASCADE'), 
        nullable=True
    )
    user: Mapped[Optional['UserModel']] = relationship(back_populates='configuration')

    enabled_globally: Mapped[bool] = mapped_column(nullable=False, default=False)

The check logic is beautifully simple:

def check_ff(self, key: ConfigKey) -> bool:
    if key in self.feature_flags:
        flag = self.feature_flags[key]

        # User-specific flag overrides global setting
        if flag.user_id is not None:
            return flag.enabled

        # Global flag
        return flag.enabled_globally

    return False

That's it. User-specific flags override global ones. Clean precedence, predictable behavior.

🎉 What I Actually Built (The Honest Truth)

Let me be real with you - I called it "feature flags" but I mostly built a dynamic configuration system.

Here are the 3 configs I actually use in production:

MINIMUM_TIME_REQUIRED_BEFORE_BOOKING_IN_HOURS - Controls booking lead time. We have it 24h but for testing it can be 0.
CAPTIONER_HOURLY_RATE_MIN - Minimum allowed hourly rate
CAPTIONER_HOURLY_RATE_MAX - Maximum allowed hourly rate

These aren't boolean flags - they're typed configuration values that I can change without deployments.

# In my booking policy
min_booking_time = now + timedelta(
    hours=Current.context.config.minimum_time_required_before_booking_in_hours
)

# In rate validation  
if not (config.captioner_hourly_rate_min <= rate <= config.captioner_hourly_rate_max):
    raise ValidationError("Rate outside allowed range")

💭 Lessons & What's Next

Build vs Buy: Started custom because it's faster and not that much to implement.

The Missing TTL Problem: Here's something I should have implemented from day one - automatic flag expiration. Feature flags are meant to be temporary. Without TTL, they become permanent technical debt. You end up with dozens of old flags cluttering your codebase and database.

In a real feature flag system, every flag should have an expiration date. When flags expire, they should either:

Auto-disable and alert the team
Force a code cleanup decision
Remove themselves entirely

This prevents the "flag graveyard" problem where you have 50+ flags and nobody remembers what half of them do.

What I'd Add Next:

TTL implementation with automated cleanup
Rename to "dynamic configuration with feature toggling"

🤔 Your Turn

What's your configuration management looking like? Are you still doing deployments to change a single value? Or have you built something similar?

I'm curious about your approach to user-specific configuration and how you handle the performance implications.

Drop a comment - I'd love to hear about your feature flag wins and disasters.

Want more posts like this? I write about practical software architecture and the real problems we solve in production. Check out my post on building event systems or fixing property drilling in FastAPI.

Stop Property Drilling in FastAPI: Use Request-Level Globals

Kuba — Sun, 21 Sep 2025 12:37:47 +0000

So in my application, I was very happy with the pattern I figured out. I knew it had some flaws like property drilling through the whole application and through multiple layers starting basically from controllers in FastAPI. I was creating the context object for the user on the route level, the same for the database session etc. I was basically passing those down through all of the layers, down to policy in domain layer just to answer "Am I allowed to do something or not".

So my code looked like this..

Before: Property drilling nightmare 🙄

💡

Notice how context is passed through multiple layers just to be used in the final policy check. This is classic property drilling - passing props through components that don't need them, just so deeply nested components can access them.

# controller.py
@router.post("/projects")
async def create_project(
    data: ProjectRequest,
    db: Session = Depends(get_db),
    context: Context = Depends(get_context)
):
    return await project_service.create_project(data, context)

# services/project_service.py
async def create_project(data: ProjectRequest, context: Context):
    # Do some business logic
    await can_create_project(context).unwrap()  # <-- this throws exception if forbidden
    return await project_repository.create(data, context)

async def can_create_project(context: Context) -> Result[bool, BaseError]:
    if not await project_policy.check_create_permission(context):
        return Result.err(
            error=BadRequestError(
                type=ErrorType.PROJECT_CREATION_FORBIDDEN,
                description='User lacks permission to create projects',
            )
        )
    return Result.ok(True)

# repositories/project_repository.py
async def create(data: ProjectRequest, context: Context):
    # Still dragging context around...
    project = Project(**data.dict())
    context.db.add(project)
    # More drilling!
    await audit_log.log_creation(project, context)  
    await event_bus.publish(ProjectCreated(project, context))
    return project

# policies/project_policy.py
class ProjectPolicy:
    def check_create_permission(self, context: Context):
        # FINALLY! We actually use the context here
        if context.user.role == "admin":
            return True
        if context.user.organization.plan == "enterprise":
            return True
        return False

The Inspiration

And then somehow I bumped into the DHH playlist on YouTube about Writing Software Well. He mentioned globals and actually the code was from Basecamp and it was in Ruby. But he showed a very important concept of Current in Ruby.

Just a disclaimer: I don't know how exactly the current looked like in Ruby and how it works. But I tried to replicate the same in my application so that I can remove a bunch of code and basically make testing easier.

The Solution

Here's the thing. The idea is that on the very beginning of the request, because we are thinking about the whole concept on the request level, not global as in the whole runtime, but only on the request level. Because a single request can be basically assigned to a single user.

So the idea is that we're creating something that is called context variable in Python on the very beginning of the request.

# context.py
from contextvars import ContextVar
from typing import Optional
from pydantic import BaseModel
from fastapi import Request

class Context(BaseModel):
    """
    Our 'current' context - similar to Ruby's Current pattern.
    Contains authenticated user and commonly used references.
    """
    issuer_id: str
    issuer_role: UserRole
    issuer_email: str
    issuer_status: UserStatus
    organization_id: Optional[str] = None
    organization_name: Optional[str] = None
    organization_member_role: Optional[OrganizationMemberRole] = None
    organization_member_status: Optional[OrganizationMemberStatus] = None
    feature_flags: dict[ConfigKey, FeatureFlagConfig] = {}
    config: UserConfig
    client_info: dict = {}

    def is_admin(self) -> bool:
        return self.issuer_role == UserRole.ADMIN

    def has_feature_flag(self, key: ConfigKey) -> bool:
        return self.config.check_ff(key)

    def verify_is_admin(self) -> None:
        if self.issuer_role != UserRole.ADMIN:
            raise ForbiddenError(
                type=ErrorType.UNAUTHORIZED,
                description='You are not authorized to access this resource',
            )

    @classmethod
    def for_user(cls, user: UserSchema) -> Self:
        organization_id = user.organization.id if user.organization else None
        organization_member_role = user.organization.role if user.organization else None
        return cls(
            issuer_id=user.id,
            issuer_role=user.role,
            issuer_status=user.status,
            issuer_email=user.email,
            organization_id=organization_id,
            organization_name=user.organization.name if user.organization else None,
            organization_member_role=organization_member_role,
            config=UserConfig(),
        )

class _ContextAttributes:
    _context: ContextVar[Optional[Context]] = ContextVar('current_context', default=None)

    @property
    def context(self) -> Context:
        ctx = self._context.get()
        if ctx is None:
            raise RuntimeError('No context available - not in request scope')
        return ctx

    @property
    def user_id(self) -> str:
        return self.context.issuer_id

    def set(self, context: Context):
        self._context.set(context)

    def clear(self):
        self._context.set(None)

# The global Current instance
Current = _ContextAttributes()

And this whole class, the instance of it, is set as a context variable in Python. And why context variable? If we use async FastAPI, the context variables are thread-safe. So if the same instance will asynchronously pick up another process, another request, we will not override this global. That's why.

Context variables in Python are specifically designed for async/concurrent code. Each async task gets its own copy of the context, preventing race conditions and data leakage between requests. The Current singleton pattern wraps the ContextVar to provide a clean interface.

Now let's see how this simplifies our code:

After: Clean and simple 🎉

# middleware.py
from fastapi import Request
from starlette.middleware.base import BaseHTTPMiddleware

class CurrentContextMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        user = await get_current_user_from_request(request)
        context = Context.for_user(user)
        context.set_client_info(request)

        Current.set(context)

        try:
            response = await call_next(request)
            return response
        finally:
            Current.clear()

# controller.py
@router.post("/projects")
async def create_project(data: ProjectRequest):
    return await project_service.create_project(data)

# services/project_service.py
async def create_project(data: ProjectRequest):
    await can_create_project().unwrap()  # Still throws if forbidden
    return await project_repository.create(data)

async def can_create_project() -> Result[bool, BaseError]:
    if not await project_policy.check_create_permission():
        return Result.err(
            error=BadRequestError(
                type=ErrorType.PROJECT_CREATION_FORBIDDEN,
                description='User lacks permission to create projects',
            )
        )
    return Result.ok(True)

# repositories/project_repository.py
async def create(data: ProjectRequest):
    context = Current.context

    project = Project(**data.dict())
    context.db_session.add(project)

    await audit_log.log_creation(project)  
    await event_bus.publish(ProjectCreated(project))
    return project

# policies/project_policy.py
class ProjectPolicy:
    def check_create_permission(self):
        # Just access the current context when needed!
        context = Current.context

        if context.is_admin():
            return True
        if context.has_feature_flag(ConfigKey.ENTERPRISE_PROJECTS):
            return True
        if context.organization_member_role == OrganizationMemberRole.OWNER:
            return True
        return False

The easy thing to do right now is basically to stop passing this context, because we already set it. And just try to access this when needed. When it's needed on the very end in the policies. When just basically checking who is the person who is asking. It's like 10% of the code. So we don't need to drill it down to the very bottom through all the layers.

Conclusion

After implementing this pattern:

40% less code in service and repository layers
Testing setup reduced by half - no more mock chains

And that's basically it. Simple trick that basically removed a lot of code, its maintainable, made everything easier, hiding some complexity behind some context magic.

And thanks, DHH, for this playlist. I think the whole series is worth watching.

Want to Know More?

Stay tuned for more insights and tutorials! Visit My Blog 🤖

Pydantic Query Params: Handling Comma-Separated Lists with Enum Validation

Kuba — Tue, 01 Jul 2025 09:00:49 +0000

Problem

You're building a FastAPI endpoint that needs to filter data by multiple criteria. Your frontend sends filter parameters as comma-separated strings (because that's how query params work), but you want proper typing with enums and optional lists on the backend.

Classic scenario:


GET /api/products?status=ACTIVE,PENDING&category=ELECTRONICS,BOOKS

But Pydantic expects lists, and you want enum validation. Plus everything should be optional.

The Challenge

Standard Pydantic approach fails here:


class FilterParams(BaseModel):
    status: Optional[list[ProductStatus]] = None# Won't work with "ACTIVE,PENDING"
    category: Optional[list[str]] = None# Won't work with "ELECTRONICS,BOOKS"

The client sends "ACTIVE,PENDING" as a single string, but you need [ProductStatus.ACTIVE, ProductStatus.PENDING].

Solution

Use BeforeValidator with a custom parser that handles both string-to-list conversion and enum casting:

from typing import Annotated, Optional
from pydantic import BaseModel, BeforeValidator
from enum import StrEnum

class ProductStatus(StrEnum):
    ACTIVE = "ACTIVE"
    PENDING = "PENDING"
    INACTIVE = "INACTIVE"

def parse_comma_separated_list(enum_type: Optional[type[StrEnum]] = None):
    def parser(v: list[str]):
# FastAPI wraps single values in lists
        value = v[0] if v else None
        if value is None:
            return None

        items = value.split(',')

# Cast to enum if provided
        if enum_type:
            return [enum_type(item) for item in items]

        return items

    return parser

class ProductFilterParams(BaseModel):
    status: Annotated[
        Optional[list[ProductStatus]],
        BeforeValidator(parse_comma_separated_list(ProductStatus))
    ] = None

    category: Annotated[
        Optional[list[str]],
        BeforeValidator(parse_comma_separated_list())
    ] = None

@router.get('/products')
async def list_products(
    filters: Annotated[ProductFilterParams, Query()],
):
# filters.status is now properly typed as list[ProductStatus] or None# filters.category is list[str] or None
    return await get_products(
        status=filters.status,
        category=filters.category
    )

How It Works

FastAPI Query Parsing: FastAPI automatically wraps query param values in lists
BeforeValidator: Runs before Pydantic's standard validation
String Splitting: Takes the first item from the list (the comma-separated string) and splits it
Enum Casting: If enum type provided, casts each item to the enum
Type Safety: Final result is properly typed for your business logic

My Experience

This hit me when refactoring an existing API. The frontend was using DiceUI filters that send multiple values as comma-separated strings.

First attempt was parsing directly in each model - messy and not reusable. Every endpoint would need its own parsing logic.

After about 2 hours of digging through Pydantic docs, I found BeforeValidator. Perfect fit - handles the transformation before validation, keeps models clean, and works everywhere.

The beauty is writing minimal code that solves the problem once and reuses everywhere.

Why This Approach?

Type Safety: Full enum validation and IDE support
Optional by Design: Handles missing params gracefully
Reusable: Works with any enum or plain strings
Clean API: Business logic gets properly typed data

The BeforeValidator pattern is perfect for these "format transformation + validation" scenarios.

That's it! Clean, reusable, and type-safe query param handling. If this helped you out, drop a like or share your own Pydantic tricks in the comments!

How I Fixed My Event Bus Before It Could Lose Money

Kuba — Tue, 17 Jun 2025 19:51:26 +0000

Remember that time when you thought you built something clever, only to discover it had fundamental flaws? Yeah, that's exactly what happened with my async event bus.

So I’ve made this post a while ago. Simple Yet Powerful: Building an In-Memory Async Event Bus in Python. And now I’ve changed my mind. I’ve found some serious problems while debugging the app. And I learn something?

Approach described in this post have some flaws. And we need to talk about them to be more aware about what we are doing.

Ofc. I encourage you to read this but short TLDR.

We have Fastapi server with async event bus.
Something important in our app happened → EventBus.publish(event) and forget.
We return the response to the user and we don’t care if handler succeed or not.
fire, forget and don’t care

For more context I will add that this bus was implemented with sync SQLAlchemy engine. Which is somehow connected also to all the problems I had and how I found them. The second problem was (And here's where it gets interesting…) the scope of the db transaction.

To the point. I will give you a few scenarios that my system was ✨supporting✨.

There was a meeting between two people. Person A did cancel the meeting and notification system failed. Person B is still waiting cuz there was no notification. Risk? Lets say moderate.
Someone was in onboarding process. Based on the role he choose, proper stripe account should be created. If I had error in stripe request, This person never got his account onboarded cuz there was no retry to create this account. Risk? A bit higher?
Person A requested additional payment from Person B. Person B agreed (EventBus.publish(dispute.resolved)). Stripe failed because there were funds missing on a card. Risk? 🚨 Money == high risk 🚨

And other problems. I don’t want to go into this domain details but I’ve noticed that as soon as we leave the ‘happy path’ the problems appears. And they are all the same:

We are loosing events.
We have wrong transaction scope.

So the refactor started…

First I’ve tackled the scope of transaction. I’ve moved the scope from single operation in repo to be a http request scope. So the commit was on the very end. Only when ALL operations succeeded.
Then I’ve reworked all methods to be async and switched the SQLAlchemy to be async engine. Still I don’t know why I initially do it sync. 🤷 There was so many problems I’ve found here with lazy loading to be fixed. Basically I found that I was lazy loading half of my app.

Ok, at this point we’ve tackled the wrong transaction scope. Right now if I choose the role of the user, related to stripe connected account, and the second part will fail, the role setup will fail also. And just to be clear. This was the design decision to avoid weird states.

user clicks → Role123 → Response: Ups! we couldn’t setup your account correctly. Please try again. ****

We don’t have any leftovers because we were in single transaction scope and there was a rollback because of exception. This was something I wanted.

But now, we not only can loose events but also we are publishing events we shouldn’t!

Look here. With initial event bus.

---> transation.open( )

    user.set_role(SUPERHUMAN)
    event_bus.publish(user_role_updated) -> has async consequences 
    stripe.create_connected_account(user) -> Throw Exception  

---> transaction.rollback( )

We’ve did a rollback but event was published. 😨

📤 Outbox Pattern

At this point we have two problems:

we might loose events if processing them will fail.
we might publish and process unwanted events.

So the solution is basically to leverage the fact of the transaction! Instead of throwing events directly to be processed to second thread, we can just… save them 🤷.

Come here I will show you my outbox. Just a simple table that stores events:

class EventOutboxModel(BaseModel):
    __tablename__ = 'events_outbox'

    class EventStatus(enum.StrEnum):
        PENDING = 'pending'
        PROCESSED = 'processed'
        FAILED = 'failed'

    event_type: Mapped[str] = mapped_column(nullable=False)
    entity_id: Mapped[str] = mapped_column(nullable=False)
    payload: Mapped[dict[str, Any]] = mapped_column(nullable=False)
    status: Mapped[EventStatus] = mapped_column(nullable=False, default=EventStatus.PENDING)
    user_id: Mapped[Optional[str]] = mapped_column(ForeignKey('users.id', ondelete='CASCADE'), nullable=True)

The fix was simple but non-obvious. Instead of firing events immediately, I started storing them:. So now what we need to change is EventBus.publish(event). Previously it was like this:

@dataclass
class Event:
    event_type: str

class SimpleEventBus:
    def __init__(self):
        self._handlers: Dict[str, List[Callable]] = {}
        self._executor = ThreadPoolExecutor(max_workers=4)

    def publish(self, event: Event) -> None:
        if event.event_type in self._handlers:
            for handler in self._handlers[event.event_type]:
                # Fire-and-forget execution
                self._executor.submit(handler, event)

What I needed to change is the session I was passing from the controller downstream.

from sqlalchemy.ext.asyncio import AsyncSession

@dataclass
class Event:
    event_type: str

class EventBus:
    async def publish(self, event: Event, db: AsyncSession) -> None:
        await self.outbox_repo.save(event, db)

Lets look one more time here now

---> transation.open( )

    user.set_role(SUPERHUMAN)
    event_bus.publis(user_role_updated, transaction) -> not yet published 
    stripe.create_connected_account(user) -> Throw Exception  

---> transaction.rollback( )

At this point events where not written to the outbox!

What does it mean? We are not publishing unwanted events. No consequences that we’ve failed after publish. Also we store ALL the events that we want to publish. So even if the event handler will fail, we can still process them one more time before switching their status to PROCESSED.

✅ We have wrong transaction scope. Leftovers after failed operations.
✅ We might loose events if processing them will fail.
✅ We might publish and process unwanted events.

Worker

Obviously there is missing part about the worker. So we have this table of events to be processed. Periodically we can fetch them and ask EventBus to point the correct event handler by event_type.
That’s it. I’ve used FastApi background task for it.

async def _loop():
    while True:
        try:
            events_count = await EventBus.flush_outbox()
        except Exception as exc:
            _logger.error(f'outbox loop error: {exc}')
        await asyncio.sleep(0.2) # Check every 200ms - fast enough for real-time feel"

@asynccontextmanager
async def lifespan(app: FastAPI):
    asyncio.create_task(_loop(), name='outbox-dispatcher')
    yield

@classmethod
async def flush_outbox(cls, batch_size: int = 100):
    count = 0
    events_to_process = []
    async with SessionLocalAsync() as session:
        events_to_process = await cls._repo.fetch_pending(session, batch_size)

    for event in events_to_process:
        for handler in cls._handlers[event.event_type]:
            try:
                async with SessionLocalAsync() as session:
                    async with session.begin():
                        await handler(e, session)
                        await cls.repo.mark_as_sent(event.id, session)
            except Exception as exc:
                cls._logger.error(f'Error handling event: {exc}')
                continue

Worker monitoring

Also one more pros I would say of this approach. With big traffic we can now monitor if we are falling behind with event processing. Look here.

# Simple monitoring query I use:
SELECT COUNT(*) FROM events_outbox WHERE status = 'pending';
# If this keeps growing, something's broken

I throw this into a simple dashboard. If I see the number climbing, I know either:

My handlers are too slow
I need more workers
Something's broken in the processing loop

Before? I had no clue if events were getting lost. Now I can actually see the queue building up.

And this story more or less ends here right now. From my perspective? Easier to test, easier to compose, more predictable…

But..

🚨 Risks and problems

Now, this solution isn't perfect - let me show you what's still broken. I can see two of them right now.

Two workers picks the same event to process. So lets say you are scaling up and suddenly two processes can pick the same event to process and publish. That means we have At least one delivery. Maybe more?

Quick fix I can think of is to create some kind of lock on events we are processing. But this is not covered here.
What if we have multiple handlers of the same events. event.status=PROCESSES means who processed? First? Second? All of them?

And this is probably the biggest issue here. If we fail on processing in one handler and second will succeed we break At least one delivery.

From the publisher perspective (outbox) we did our job. And here comes the role of the consumer. Solution? Maybe next blog post will be about inbox pattern?

Anyway, that's how I stopped losing events before they could cost me real money. Pretty sure there are more edge cases waiting to bite me…

Want more? 🤔

Leave a like or comment and feed the algorithm! Easy for you and give me a lot of boost. Thanks!

If you want more you can check it here

Build a Policy-Based Access System in Python

Kuba — Fri, 06 Jun 2025 22:46:22 +0000

Last week, I was staring at permission checks scattered across my API codebase. You know the feeling - nested if-statements everywhere, duplicated logic, and that voice saying "there's gotta be a better way.”

I’m allowed to query /user/123 if:
- I’m user with id 123.
- I’m Admin.
- I’m Admin of organisation to which user 123 belongs to.
I’m allowed to query /meeting/123
- When I’m admin
- When I’m participant of /meeting/123
- When I’m BillingAdmin in organisation to which one of participants belongs to.

So as you see there is like a plenty of examples which are pretty simple to implement. I just want to avoid things like this across the whole codebase.

if context.issuer_id != user_id:
    raise ForbiddenError('Go away!')

One more thing for context. This was built to be applied when querying for the data. Not when performing operations. Performing actions is much more complicated and implemented in different way. Can describe this if needed.

My main requirements

I can easily compose different rules.
I can easily add this to existing codebase.
I need to fail fast. So reject query as soon as possible.

End Goal Overview

This is what I was looking for. Easily apply rules that:

this method can be accessed if user/123 is asking for himself.
this method can be accessed by Admin
this method can be accessed by Admin of the organisation that user/123 belongs to.

@require_any([
    RequireIssuerIdPolicy("user_id"),
    RequireAnyRolePolicy([UserRole.ADMIN]),
    RequireOrganizationAdminPolicy()
])
def get_user_profile(self, user_id: str, context: Context) -> UserProfile:
    return self.user_repo.get_profile(user_id)

The whole point here is that in single place we can easily add checks that require no data and require the result of the function. So the decorator must be smart enough to manage them all.

Implementation details

Lets go with implementation. Here is one nice thing. The key assumption was that across the whole app we are passing context object which is prepared in FastApi middleware based on JWT token. It has user data. So by default, for all endpoints we can check if user is actually at least active.

Decorators

def require_any(policies: list[AccessPolicy]):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            context: Context | None = kwargs.get('context')
            if not context:
                raise ValueError("Context must be provided as a 'context' keyword argument")

            if context.issuer_status in [UserStatus.DEACTIVATED, UserStatus.PENDING_APPROVAL]:
                raise ForbiddenError(ErrorType.FORBIDDEN, 'Access denied: user is deactivated or pending approval')

            post_checks = []

            # Try all pre-checks first
            pre_passed = []

            for policy in policies:
                if policy.is_post_check():
                    post_checks.append(policy)
                else:
                    try:
                        policy.check(context, kwargs)
                        pre_passed.append(policy)
                    except ForbiddenError:
                        continue

            result = func(*args, **kwargs)
            if pre_passed or not policies:
                return result

            # If none of pre passed, try post checks
            for policy in post_checks:
                try:
                    policy.check_post(context, result, kwargs)
                    return result
                except ForbiddenError:
                    continue

            raise ForbiddenError(ErrorType.FORBIDDEN, 'Access denied: no policy passed (require_any)')
        return wrapper
    return decorator

Policies

class AccessPolicy(ABC):
    def is_post_check(self) -> bool:
        return False

    @abstractmethod
    def check(self, context: Context, kwargs: dict):
        pass

    def check_post(self, context: Context, result: object, kwargs: dict):
        raise NotImplementedError('Not a post-execution policy')

class PostAccessPolicy(AccessPolicy):
    def is_post_check(self) -> bool:
        return True

    def check(self, context: Context, kwargs: dict):
        raise NotImplementedError('Post check uses check_post() instead')

class RequireIssuerIdPolicy(AccessPolicy):
    def __init__(self, user_id_arg: str = 'user_id'):
        self.user_id_arg = user_id_arg

    def check(self, context: Context, kwargs: dict):
        user_id = kwargs.get(self.user_id_arg)
        if user_id != context.issuer_id:
            raise ForbiddenError(ErrorType.FORBIDDEN, 'Access denied')

class RequireOrganizationAdminPolicy(PostAccessPolicy):
    # Check if the issuer is an organization admin and the resource is in the same organization.
    def __init__(self, organization_id_arg: str = 'organization.id'):
        self.organization_id_arg = organization_id_arg

    def check_post(self, context: Context, result: object, kwargs: dict):
        resource_organization_id = result
        for step in self.organization_id_arg.split('.'):
            resource_organization_id = getattr(resource_organization_id, step)

        if context.organization_member_role != OrganizationMemberRole.ADMIN:
                    raise ForbiddenError(ErrorType.FORBIDDEN, 'Access denied: not organization admin')
                if resource_organization_id != context.organization_id:
                    raise ForbiddenError(ErrorType.FORBIDDEN, 'Access denied: different organization')

class RequireAnyRolePolicy(AccessPolicy):
    def __init__(self, roles: list[UserRole]):
        self.roles = roles

    def check(self, context: Context, kwargs: dict):
        if context.issuer_role not in self.roles:
            raise ForbiddenError(ErrorType.FORBIDDEN, f'Access denied: requires one of {self.roles}')

Why Post-Execution Checks And Pre-Execution Checks?

Pre-execution checks run before your function executes. They will let you reject unauthorized requests before expensive database queries, saving both time and resources.. These are perfect for simple validations like:

"Is user an admin?"
"Does the user ID match the requester?"

Post-execution checks run after fetching the resource. Some access decisions require data only available after fetching the resource. For example:

"Is the user a participant in this specific meeting?"
"Does the user belong to the same organization as this booking?"

Problems Solved

This system tackles several common pain points in API access control:

Scattered Permission Logic: No more if-statements buried throughout your business logic
Code Duplication: Write once, reuse everywhere - no copy-pasting permission checks
Resource-Dependent Access: Handle permissions that depend on fetched data (like "is user a meeting participant?")
Maintainability: Easy to modify, test, and extend without touching business logic

Want to Know More?

Stay tuned for more insights and tutorials! Visit My Blog 🤖

I want to build a startup - how I use AI and my first ideas.

Kuba — Sun, 02 Mar 2025 12:54:14 +0000

Ok, I think it’s time to officially admit that I want to build something and I want to make it public! Time to go out of the basement.

In general I this post is a bit of shift I want to do from writing only about “how to do xyz” to writing about “what I’m currently doing”. Seems a bit challenging to me but let’s give it a go.

A bit of context

Right now, everything is about AI, how easy it is to make software, how fast we can ship etc. Since I’ve always was thinking about building something I thought its time to stop thinking and start doing.

A month has passed, and I have 3 ideas in progress.

But more about them later.

The AI Role

Obviously I’m using AI for them a lot. And I have some thoughts. We have awesome tools. Cursor, Claude, GPT etc. we can count them all but its not the point.

They bring some solutions together with some problems (that I think can be avoided in long term).

Maintain or Ship MVP?

That’s the first question I would ask. And It’s not about using AI or not. For me it’s more about how caution I should be when using them.

Will give you an example. Initial views for www.mydancedna.com were files 1k lines or more. AI Generated with Cursor and Claude. I’ve setup the project with react and shadcn components and I was happily pointing to docs and building.

And here comes the tradeoff for not paying enough attention. The code was almost impossible to maintain. The moment I’ve started discovering some bugs or requesting very little changes, it was harder to achieve the right because of too big context.

The solution: from the very beginning. Try to be explicit about the code structure, how split components, how to not only build the solution but also how to build the codebase itself. I found the .cursorrules very helpful.
Right now I’m also exploring the *.mdc files for the rules. Maybe will write about them more later.

How much I need to validate?

Honestly? Idk. Didn’t validate successfully anything so far 🤷.

But! I’m trying to prevent myself from building the whole system before I will show it to somebody. I think people in general are able to imagine something. Maybe also imaging something you are not even expecting them to do.

What I’m saying here is that I’m trying to show them only mocked frontend views right now. Ofc. I have some backend written, there are some http calls flying back and forth. But the current strategy is something like:

generate as much ‘clickable’ views you can to show the idea.
give it to people to click.
gather feedback.

Actually.. if you want to click it a bit and give any feedback or thoughts? I would really appreciate that 🙏.

🌐 here is the link: https://www.mydancedna.com/

📜 here is the form: My Dance Dna Feedback

How I fill missing positions in my startup.

I’ll be honest. There is a planty of things I have no idea how to do. Starting from designing the frontend, the UX, writing a good copy, product design, talking to people etc. I’m good at backend stuff (period).

That means that there is a lot of open positions in my little startups 😅.

And yes, so far I’m just trying to fill them with AI tools.

Missing the correct scope for MVP: write a prompt for product designer. Talk with him what could be the best scope. Document it and move on.

Missing landing page: write a prompt for UX/UI designer. Talk with him what should be included based on your documentation. Document it and use Cursor.

Missing XYZ: write a prompt for ABC. Talki with him … Document it… and move on.

This is awesome way of doing things so far 👆. Looking at how much caution I need to be when it comes to backend stuff, I assume there is a lot of pros that will tell me why it can be bad. But so far? It’s enough.

The key takeaway here would be context again. The part where I say “document it”. Those “ai hires” need to share some context. It’s pain in the ass to explain each time what the project is about. That’s why I’m doing more documentation in my personal projects then professional ones. lol. The better common understanding of what they need to achieve the better results they will produce.

What’s next?

Validation, Validation, Validation. I really want to build and I’m forcing myself to stop before next steps. At least that’s what internet gurus are advising.

So one little small reminder:

🌐 here is the link: https://www.mydancedna.com/

📜 here is the form: My Dance Dna Feedback

Other 2 ideas

I said that I’ve 3 ideas in progress. MyDanceDna is the biggest one. But there are two other “tools that I would like to have”

db client with:
- graph view based on foreign keys. I think that would be useful. Jumping into new project and being able to see visually those relations and dependencies.
- filtering on this graph view. This is what I’m missing sometimes when debugging. Lets say in easy example you have users table, orders table and products table. And In this graph view I have filtered by user, so when I click on this table relations I have already filters applied based on foreign keys. I think that would be useful for devs. Thoughts?
visual api tests builder:

I found that keeping good openapi schema is sooooo helpful. So I’m trying to build this tools where:
- You can upload the schema and see all endpoints and their structure.
- Visually build flows and business processes.
- SAVE and RUN them from local machine. Something like e2e tests.

Any thoughts about those two? Please put a comment if you find them interesting!

Want to Know More?

Stay tuned for more insights! Visit My Blog 🤖

Simple Yet Powerful: Building an In-Memory Async Event Bus in Python

Kuba — Fri, 31 Jan 2025 08:51:16 +0000

So I've been working on this project where different parts of the system needed to talk to each other. You know, the usual stuff - user signs up, order gets updated, notifications need to be sent... The typical advice? "Just use RabbitMQ or Redis!" But hmm... that felt like bringing a tank to a knife fight 🤷‍♂️. Especially that time to the market matters and I was the only dev there. So I wanted to be able to move fast.

The Problem

I needed something simple:

Async event handling (fire and forget)
No fancy setup
Easy to test
Keeps different parts of the code separate

The Solution: Let's Build Our Own

Here's what I came up with:

from concurrent.futures import ThreadPoolExecutor
from typing import Callable, Dict, List
from dataclasses import dataclass

@dataclass
class Event:
    event_type: str

class SimpleEventBus:
    def __init__(self):
        self._handlers: Dict[str, List[Callable]] = {}
        self._executor = ThreadPoolExecutor(max_workers=4)

    def publish(self, event: Event) -> None:
        if event.event_type in self._handlers:
            for handler in self._handlers[event.event_type]:
                # Fire-and-forget execution
                self._executor.submit(handler, event)

    def subscribe(self, event_type: str, handler: Callable) -> None:
        if event_type not in self._handlers:
            self._handlers[event_type] = []
        self._handlers[event_type].append(handler)

So How Does Python Threading Actually Work?

This one is crucial to understand what's happening here.

The Global Interpreter Lock (GIL)

First thing you need to know - Python has this thing called the GIL. It's basically Python's way of managing memory safely in a multi-threaded environment. Here's how it actually works:

Only one thread can execute Python code at a time
When a thread starts doing I/O operations (like HTTP requests, file operations, or DB calls), it releases the GIL
The GIL then becomes available for other threads to use
This is why I/O heavy tasks can actually run in parallel!

A small detail that many don't know: When you're doing CPU work (like calculations), you're stuck with one thread at a time. But for I/O work (which is what most event handlers do), Python can run multiple threads effectively in parallel. This is perfect for our event bus since most handlers are doing I/O anyway!

💣 Quick reminder: if you find this useful, please give it a thumbs up! Your support helps me create more content.

CPU vs I/O Tasks

So here's the cool part:

CPU tasks (like heavy calculations) - one at a time, no way around it
I/O tasks (like making HTTP calls or reading files) - Python's like "go ahead, do your thing!" and lets multiple threads work

Check out what this looks like in practice. Here's how things flow when we publish an event:

Check here for better rendered graphs

sequenceDiagram
    participant C as Client Code
    participant EB as Event Bus
    participant TP as Thread Pool
    participant W1 as Worker 1
    participant W2 as Worker 2
    participant GIL as Python GIL

    C->>EB: publish(UserCreatedEvent)
    activate EB
    EB->>TP: submit(handler1)
    EB->>TP: submit(handler2)
    deactivate EB
    Note over EB: Returns immediately!

    TP->>W1: execute(handler1)
    TP->>W2: execute(handler2)

    W1->>GIL: request GIL
    GIL->>W1: grant GIL
    Note over W1: Execute I/O operation<br>(send email)
    W1-->>GIL: release GIL during I/O

    W2->>GIL: request GIL
    GIL->>W2: grant GIL
    Note over W2: Execute I/O operation<br>(write to log)
    W2-->>GIL: release GIL during I/O

When one thread is waiting for I/O (like sending an email), it lets go of the GIL and other threads can do their thing. That's why this works so well for our event handlers - they're mostly doing I/O stuff anyway!

And how this looks like on CPU:

Check here for better rendered graphs

gantt
    title Thread Execution Patterns with GIL
    dateFormat X
    axisFormat %s

    section CPU-Bound Tasks
    Thread 1 (CPU Calculation)    :crit, cpu1, 0,  5
    Waiting for GIL              :wait1, 5,  6
    Thread 2 (CPU Calculation)    :crit, cpu2, 6,  11
    Waiting for GIL              :wait2, 11, 12
    Thread 3 (CPU Calculation)    :crit, cpu3, 12, 17

    section I/O-Bound Tasks
    Thread 4 (DB Query)          :active, io1, 0,  8
    Thread 5 (HTTP Request)      :active, io2, 2,  10
    Thread 6 (File Write)        :active, io3, 4,  12
    Thread 7 (Email Send)        :active, io4, 6,  14
    Thread 8 (Redis Operation)   :active, io5, 8,  16

CPU-Bound Section:
- Notice how CPU tasks (red) NEVER overlap
- Each thread must wait for the GIL before proceeding
- Small gaps between tasks show GIL handoff time
I/O-Bound Section:
- Multiple I/O operations (blue) run in parallel
- DB queries, HTTP requests, file operations all overlap
- Much better throughput than CPU-bound tasks

How Does It Work?

The magic happens in these two lines:

def publish(self, event: Event) -> None:
    self._executor.submit(handler, event) # This is the key!

When you publish an event:

Your code continues running immediately
The event handlers run in the background
You don't have to wait for them to finish.

Real World Example

Let's see how this actually works on some example:

@dataclass
class OrderPlaced(Event):
    order_id: str
    user_id: str
    items: list[dict]
    total_amount: Decimal
    shipping_address: dict

bus = SimpleEventBus()

# Different parts of the system handle different responsibilities
@SimpleEventBus.handler("order_placed")
def notify_warehouse(event: OrderPlaced):
    """Warehouse needs to know about new orders to prepare shipment"""
    warehouse_client.create_shipment_order(
        order_id=event.order_id,
        items=event.items,
        address=event.shipping_address
    )

@SimpleEventBus.handler("order_placed")
def send_confirmation_email(event: OrderPlaced):
    """Customer should get order confirmation"""
    email_template = load_template("order_confirmation")
    send_email(
        to=get_user_email(event.user_id),
        subject=f"Order #{event.order_id} Confirmation",
        body=email_template.render(
            items=event.items,
            total=event.total_amount
        )
    )

@SimpleEventBus.handler("order_placed")
def update_inventory(event: OrderPlaced):
    """Update stock levels"""
    for item in event.items:
        inventory.decrease_stock(
            product_id=item['id'],
            quantity=item['quantity']
        )

@SimpleEventBus.handler("order_placed")
def trigger_analytics(event: OrderPlaced):
    """Track business metrics"""
    analytics.track_sale(
        order_id=event.order_id,
        amount=event.total_amount,
        user_id=event.user_id
    )

# In your order service, you just:
class OrderService:
    def place_order(self, user_id: str, items: list, address: dict) -> str:
        # Core business logic
        order = Order.create(
            user_id=user_id,
            items=items,
            shipping_address=address
        )

        # Just publish the event - everything else happens automatically!
        bus.publish(OrderPlaced(
            order_id=order.id,
            user_id=user_id,
            items=items,
            total_amount=order.total,
            shipping_address=address
        ))

        return order.id

The beauty of this approach is that:

OrderService doesn't need to know about emails, warehouses, or analytics
Adding new functionality is just adding new handlers
Each handler can fail independently without affecting others
Easy to test each piece in isolation

For example, adding loyalty points later is as simple as:

@SimpleEventBus.handler("order_placed")
def award_loyalty_points(event: OrderPlaced):
    points = calculate_points(event.total_amount)
    loyalty_service.add_points(
        user_id=event.user_id,
        points=points,
        reason=f"Order #{event.order_id}"
    )

No need to modify the existing order code - just add the new handler, test it separately and mark this Jira ticket as done!

Testing Is Super Easy

Here's a nice trick for testing - create a synchronous executor. I’m using this one to be able to assert on more e2e tests where I want to make sure that something has happened in the background.

class SyncExecutor(Executor):
    """Runs everything in the same thread - perfect for testing"""
    def submit(self, fn: Callable[P, T], /, *args: P.args, **kwargs: P.kwargs) -> Future[T]:
        result = fn(*args, **kwargs)
        future: Future[T] = Future()
        future.set_result(result)
        return future

def test_user_created_handlers():
    # Simple list to track handler calls
    handled_events = []

    def test_handler(event: UserCreated):
        handled_events.append(event)

    # Setup bus with sync executor
    bus = SimpleEventBus(executor=SyncExecutor())
    bus.subscribe("user_created", test_handler)

    # Publish test event
    test_event = UserCreated("test-id", "test@example.com")
    bus.publish(test_event)

    # With SyncExecutor, we can check immediately
    assert len(handled_events) == 1
    assert handled_events[0].user_id == "test-id"

Thread Safety Considerations

When you're working with threads, there are a few gotchas you need to watch out for. Let me show you some real problems I've hit and how to solve them.

1. Handler State

The most common issue? Shared state in handlers. Look at this:

# DON'T DO THIS
total_orders = 0

@SimpleEventBus.handler("order_placed")
def track_orders(event: OrderPlaced):
    global total_orders
    total_orders += 1  # 💥 Race condition!

Instead, either use thread-safe counters or better yet, avoid shared state:

from threading import Lock

class OrderTracker:
    def __init__(self):
        self._count = 0
        self._lock = Lock()

    def increment(self):
        with self._lock: # this is like public toilet. Only one person at a time. 
            self._count += 1
            return self._count

# Better approach - use the tracker
tracker = OrderTracker()

@SimpleEventBus.handler("order_placed")
def track_orders(event: OrderPlaced):
    new_count = tracker.increment()
    logger.info(f"Processed order {event.order_id}. Total: {new_count}")

2. Resource Management

Another classic problem - sharing resources between handlers. Here's what not to do:

# DON'T DO THIS - shared connection between threads
db_connection = create_db_connection()

@SimpleEventBus.handler("order_placed")
def save_order(event: OrderPlaced):
    db_connection.execute(  # 💥 Not thread-safe!
        "INSERT INTO orders VALUES (?)", 
        [event.order_id]
    )

Instead, manage resources per handler call:

# DO THIS - each handler gets its own connection
@SimpleEventBus.handler("order_placed")
def save_order(event: OrderPlaced):
    with create_db_connection() as conn:  # Thread-safe! 
        conn.execute(
            "INSERT INTO orders VALUES (?)",
            [event.order_id]
        )

The Main Rules

Never share mutable state between handlers
Use thread-safe data structures when you must share
Manage resources (DB connections, API clients) per handler call
When in doubt, use Queue from the standard library

Remember: The event bus takes care of most thread safety for you. These issues usually come up in the handlers themselves. Keep your handlers simple and stateless, and you'll avoid most problems! 🎯

When Should You Use This?

This simple approach is perfect when:

You don't need events to survive app restarts
You're not handling thousands of events per second
You want to avoid complex infrastructure
You need something easy to test

But maybe look at RabbitMQ or Kafka when you need:

Events that persist
Guaranteed delivery
Super high throughput
Communication between different services

The Results?

We've been running this for a while now. It just works. No maintenance headaches, and when I need to add new events or handlers, it's straightforward. Also if needed it will be pretty easy to move to more advanced solutions like Kafka etc. cuz the whole infrastructure is already in place throwing events.

Remember: Sometimes the simple solution is the right solution 🎯

💣 Does this post help you? Please spam the like button! Your support is awesome. Thanks!

Want to Know More?
Stay tuned for more insights and tutorials! Visit My Blog 🤖

Local Development Server for AWS SAM Lambda Projects

Kuba — Sat, 28 Sep 2024 12:32:22 +0000

Right now I’m working on a project where REST API is built using AWS lambdas as request handlers. The whole thing uses AWS SAM to define lambdas, layers and connect it to Api Gateway in nice template.yaml file.

The Problem

Testing this API locally isn't as straightforward as with other frameworks. While AWS provides sam local commands to build Docker images that host lambdas (which better replicate the Lambda environment), I found this approach too heavy for quick iterations during development.

The Solution

I wanted a way to:

Quickly test my business logic and data validations
Provide a local server for frontend developers to test against
Avoid the overhead of rebuilding Docker images for every change

So, I created a script to address these needs. 🤷‍♂️

TL;DR: Check out server_local.py in this GitHub repository.

Key Benefits

Quick Setup: Spins up a local Flask server that maps your API Gateway routes to Flask routes.
Direct Execution: Triggers the Python function (Lambda handler) directly, without Docker overhead.
Hot Reload: Changes are reflected immediately, shortening the development feedback loop.

This example builds on the "Hello World" project from sam init, with server_local.py and its requirements added to enable local development.

Reading the SAM Template

What I’m doing here is I’m reading the template.yaml first since there is current definition of my infrastructure and all the lambdas.

All the code we need to create a dict definition is this. To handle functions specific for SAM template I’ve added some constructors to CloudFormationLoader. It now can support Ref as reference to another object, Sub as method to substitute and GetAtt to get attributes. I think we can add more logic here but right now this was totally sufficient to make it work.

import os
from typing import Any, Dict
import yaml


class CloudFormationLoader(yaml.SafeLoader):
    def __init__(self, stream):
        self._root = os.path.split(stream.name)[0]  # type: ignore
        super(CloudFormationLoader, self).__init__(stream)

    def include(self, node):
        filename = os.path.join(self._root, self.construct_scalar(node))  # type: ignore
        with open(filename, "r") as f:
            return yaml.load(f, CloudFormationLoader)


def construct_getatt(loader, node):
    if isinstance(node, yaml.ScalarNode):
        return {"Fn::GetAtt": loader.construct_scalar(node).split(".")}
    elif isinstance(node, yaml.SequenceNode):
        return {"Fn::GetAtt": loader.construct_sequence(node)}
    else:
        raise yaml.constructor.ConstructorError(
            None, None, f"Unexpected node type for !GetAtt: {type(node)}", node.start_mark
        )


CloudFormationLoader.add_constructor(
    "!Ref", lambda loader, node: {"Ref": loader.construct_scalar(node)}  # type: ignore
)
CloudFormationLoader.add_constructor(
    "!Sub", lambda loader, node: {"Fn::Sub": loader.construct_scalar(node)}  # type: ignore
)
CloudFormationLoader.add_constructor("!GetAtt", construct_getatt)


def load_template() -> Dict[str, Any]:
    with open("template.yaml", "r") as file:
        return yaml.load(file, Loader=CloudFormationLoader)

And this produces json like this:

{
   "AWSTemplateFormatVersion":"2010-09-09",
   "Transform":"AWS::Serverless-2016-10-31",
   "Description":"sam-app\nSample SAM Template for sam-app\n",
   "Globals":{
      "Function":{
         "Timeout":3,
         "MemorySize":128,
         "LoggingConfig":{
            "LogFormat":"JSON"
         }
      }
   },
   "Resources":{
      "HelloWorldFunction":{
         "Type":"AWS::Serverless::Function",
         "Properties":{
            "CodeUri":"hello_world/",
            "Handler":"app.lambda_handler",
            "Runtime":"python3.9",
            "Architectures":[
               "x86_64"
            ],
            "Events":{
               "HelloWorld":{
                  "Type":"Api",
                  "Properties":{
                     "Path":"/hello",
                     "Method":"get"
                  }
               }
            }
         }
      }
   },
   "Outputs":{
      "HelloWorldApi":{
         "Description":"API Gateway endpoint URL for Prod stage for Hello World function",
         "Value":{
            "Fn::Sub":"https://${ServerlessRestApi}.execute-api.${AWS::Region}.amazonaws.com/Prod/hello/"
         }
      },
      "HelloWorldFunction":{
         "Description":"Hello World Lambda Function ARN",
         "Value":{
            "Fn::GetAtt":[
               "HelloWorldFunction",
               "Arn"
            ]
         }
      },
      "HelloWorldFunctionIamRole":{
         "Description":"Implicit IAM Role created for Hello World function",
         "Value":{
            "Fn::GetAtt":[
               "HelloWorldFunctionRole",
               "Arn"
            ]
         }
      }
   }
}

Handling Layers

Having that its easy to dynamically create Flask routes for each endpoint. But before that something extra.

In sam init helloworld app there are no layers defined. But I had this problem in my real project. To make it work properly I’ve added a function that reads layers definitions and add them to sys.path that python imports can work correctly. Check this:

def add_layers_to_path(template: Dict[str, Any]):
    """Add layers to path. Reads the template and adds the layers to the path for easier imports."""
    resources = template.get("Resources", {})
    for _, resource in resources.items():
        if resource.get("Type") == "AWS::Serverless::LayerVersion":
            layer_path = resource.get("Properties", {}).get("ContentUri")
            if layer_path:
                full_path = os.path.join(os.getcwd(), layer_path)
                if full_path not in sys.path:
                    sys.path.append(full_path)

Creating Flask Routes

In the we need to loop throughout resources and find all functions. Based on that Im creating data need for flask routes.

def export_endpoints(template: Dict[str, Any]) -> List[Dict[str, str]]:
    endpoints = []
    resources = template.get("Resources", {})
    for resource_name, resource in resources.items():
        if resource.get("Type") == "AWS::Serverless::Function":
            properties = resource.get("Properties", {})
            events = properties.get("Events", {})
            for event_name, event in events.items():
                if event.get("Type") == "Api":
                    api_props = event.get("Properties", {})
                    path = api_props.get("Path")
                    method = api_props.get("Method")
                    handler = properties.get("Handler")
                    code_uri = properties.get("CodeUri")

                    if path and method and handler and code_uri:
                        endpoints.append(
                            {
                                "path": path,
                                "method": method,
                                "handler": handler,
                                "code_uri": code_uri,
                                "resource_name": resource_name,
                            }
                        )
    return endpoints

Then next step is to use it and setup a route for each one.

def setup_routes(template: Dict[str, Any]):
    endpoints = export_endpoints(template)
    for endpoint in endpoints:
        setup_route(
            endpoint["path"],
            endpoint["method"],
            endpoint["handler"],
            endpoint["code_uri"],
            endpoint["resource_name"],
        )


def setup_route(path: str, method: str, handler: str, code_uri: str, resource_name: str):
    module_name, function_name = handler.rsplit(".", 1)
    module_path = os.path.join(code_uri, f"{module_name}.py")
    spec = importlib.util.spec_from_file_location(module_name, module_path)
    if spec is None or spec.loader is None:
        raise Exception(f"Module {module_name} not found in {code_uri}")
    module = importlib.util.module_from_spec(spec)

    spec.loader.exec_module(module)
    handler_function = getattr(module, function_name)

    path = path.replace("{", "<").replace("}", ">")

    print(f"Setting up route for [{method}] {path} with handler {resource_name}.")

    # Create a unique route handler for each Lambda function
    def create_route_handler(handler_func):
        def route_handler(*args, **kwargs):
            event = {
                "httpMethod": request.method,
                "path": request.path,
                "queryStringParameters": request.args.to_dict(),
                "headers": dict(request.headers),
                "body": request.get_data(as_text=True),
                "pathParameters": kwargs,
            }
            context = LambdaContext(resource_name)
            response = handler_func(event, context)

            try:
                api_response = APIResponse(**response)
                headers = response.get("headers", {})
                return Response(
                    api_response.body,
                    status=api_response.statusCode,
                    headers=headers,
                    mimetype="application/json",
                )
            except ValidationError as e:
                return jsonify({"error": "Invalid response format", "details": e.errors()}), 500

        return route_handler

    # Use a unique endpoint name for each route
    endpoint_name = f"{resource_name}_{method}_{path.replace('/', '_')}"
    app.add_url_rule(
        path,
        endpoint=endpoint_name,
        view_func=create_route_handler(handler_function),
        methods=[method.upper(), "OPTIONS"],
    )

And you can start your server with

if __name__ == "__main__":
    template = load_template()
    add_layers_to_path(template)
    setup_routes(template)
    app.run(debug=True, port=3000)

That’s it. The whole code available on github https://github.com/JakubSzwajka/aws-sam-lambda-local-server-python. Let me know if you find any corner case with layers etc. That can be improved or you think its worth adding something more to this. I find it very helpful.

Potential Issues

In short this works on your local environment. Keep in mind that lambdas has some memory limitations applied and cpu. In the end its good to test it in real environment. This approach should be used to just speed up development process.

If you implement this in your project, please share your insights. Did it work well for you? Any challenges you faced? Your feedback helps improve this solution for everyone.

Want to Know More?

Stay tuned for more insights and tutorials! Visit My Blog 🤖

Chat with HTMX, WebSockets and Hono

Kuba — Tue, 06 Aug 2024 22:06:47 +0000

Last week, I wrote about tweaking htmx to display instant messages. A week into using HTMX, I needed more. I wanted a better way to stream HTML from the server, using JSX components instead of plain HTML strings for better code usability.

💣 Quick reminder: if you find this useful, please give it a thumbs up! Your support helps me create more content.

Tools I Used:

HTMX
HTMX Websockets Extension
Hono for the backend
Websockets - client-side

The idea is simple. My Conversation component is wrapped in a div with hx-ext="ws", which connects to my backend when rendered.

export const Conversation = (props: { messages: Message[] }) => (
  <div hx-ext="ws" ws-connect="/chatroom-ws">
      <div id="conversation">
        {props.messages.reverse().map((message) => (
          <div>
            <UserMessage message={message} />
            <AssistantMessage message={message} />
          </div>
        ))}
      </div>
      <InputMessageForm />
  </div>
);

Next important thing is the InputMessageForm. Just add ws-send to the form, and it will send a message where the key is the textarea’s ID (messageInput) with its value.

export const InputMessageForm = () => (
  <form id="query-submit-form" ws-send className="relative">
    <textarea
      id="messageInput"
      name="userMessage"
      placeholder="Type your message here..."
      rows={4}
    ></textarea>
    <button type="submit">Send</button>
  </form>
);

Websockets - Server

Here’s the full code block for the Hono server. Some console logs for opening and closing connection. onMessage is where the magic happens.



get(
    '/chatroom-ws',
    upgradeWebSocket((c) => {
      return {
        onOpen: () => {
          console.log('WS Connection open');
        },
        onClose: () => {
          console.log('WS Connection closed');
        },
        onMessage: async (event, ws) => {
          const { userMessage } = JSON.parse(event.data.toString());
          console.log('Got user message', userMessage);
          const inputArea = await c.html(
            <div id="query-submit-form">
              <InputMessageForm />
            </div>,
          );
          ws.send(await inputArea.text());
          const htmlUser = await c.html(
            <div id="conversation" hx-swap-oob="beforeend">
              <UserMessage
                message={{
                  id: v4(), // some random ids used here for placeholder
                  query: userMessage,
                  completion: '',
                  conversationId: v4(),
                  toolsResponse: null,
                  createdAt: new Date(),
                  updatedAt: new Date(),
                }}
              />
            </div>,
          );
          ws.send(await htmlUser.text());
          const response = await talk(userMessage);
          const htmlAgent = await c.html(
            <div id="conversation" hx-swap-oob="beforeend">
              <AssistantMessage message={response} />
            </div>,
          );
          ws.send(await htmlAgent.text());
        },
      };
    }),
);

So the flow is:

Receive the query
Send back empty just to make it clean. There is no hx-swap-oob specified so its True by default. That means that it will find the element with id=query-submit-form and swap it.
Send back the component with the user message. Here hx-swap-oob is specified to beforeend which simply means that it will be added to existing messages.
talk → here comes your logic. I’m talking to AI assistant so making some external API calls.
Send back the component with assistant answer. The same as step 3 but the component is different.

Problems I found

Sending response back was a bit problematic since docs are hmm… not that easy to understand I think. There is even an issue created to fix this: Improve documentation for websocket extension. That helped me a lot!

So the most important thing is:

You need to send back string, that parses to html that has the same id as the thing you want to swap!

So the problem nr. 1

I accidentally sent back something like this:

JSON.stringify('<div id="someid">test 123</div>')
// '"<div id=\\"someid\\">test 123</div>"'

This is wrong. Note the ID and escape characters! Don’t stringify the string here.

The problem nr. 2

You might think you can return something and it will get swapped where you want. Not exactly. The first div is just information for HTMX on what to do. At least I understand it this way 🤔.

I’m returning html like this:

<div id="conversation" hx-swap-oob="beforeend">
      <AssistantMessage message={response} />
</div>

Only is appended inside the existing

on the client side.

End result

https://assets.super.so/c0fc84d8-fb32-4194-8758-4be657666aab/videos/c814dcd2-b9e9-4bb2-b8db-2ed9cd7819b7/lucy-chat-example.mov

💣 Does this post help you? Please spam the like button! Your support is awesome. Thanks!

Want to Know More?

Stay tuned for more insights and tutorials! Visit My Blog 🤖

Instant Message Display in Chat Apps with htmx

Kuba — Thu, 01 Aug 2024 13:22:16 +0000

Problem overview

While building my AI assistant, I needed a simple chat interface to test it locally. Building a whole React app seemed overkill, so I tried HTMX. The challenge was that the response logic took time (at least 5 calls to OpenAI API) for a single query. For better UX, I wanted the user message displayed instantly and the assistant’s response appended afterward.

Solution

TLDR: repo is here: https://github.com/JakubSzwajka/hono-htmx-chat

The idea is pretty simple:

Send query and expect only the assistant’s response in ready-to-display HTML.
Append user message to the conversation upon form submission.

For this example I’ve added setTimeout to simulate some logic.

// index.ts (so called 'backend') 

const app = new Hono()

const messages: { user: string, message: string }[] = []

app.get('/', serveStatic({ path: './static/index.html' }))

app.get('/messages', (c) => {
  return c.html(messages.map(msg => `<p><strong>${msg.user}:</strong> ${msg.message}</p>`).join(''))
})

app.post('/message', async (c) => {
  const formData = await c.req.formData()
  const message = formData.get('message') as string
  messages.push({ user: 'user', message })

  await new Promise(resolve => setTimeout(resolve, 2000))

  // Simple response logic
  const responseMessage = `Received your message: ${message}`
  messages.push({ user: 'Assistant', message: responseMessage })

  return c.html(`
    <p><strong>Assistant:</strong> ${responseMessage}</p>
  `)
})

Client logic here does the trick. Simple function that appends user query to the conversation body.

// index.html 
<!DOCTYPE html>
<html lang="en">
  <head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>Chat App</title>
    <script src="https://unpkg.com/htmx.org@1.6.1"></script>
  </head>
  <body>
    <h1>Chat App</h1>
    <div id="chat" hx-get="/messages" hx-trigger="load" hx-swap="innerHTML">
      <!-- Chat messages will be displayed here -->
    </div>
    <form id="messageForm" hx-post="/message" hx-trigger="submit"
      hx-swap="beforeend"
      hx-target="#chat"
      onsubmit="handleSubmit(event)">
      <input type="text" id="message" name="message" placeholder="Your message"
        required>
      <button type="submit">Send</button>
    </form>
  </body>
  <script>
    function handleSubmit(event) {
      event.preventDefault();
      const message = new FormData(event.target).get('message');
      const newMessage = document.createElement('p');
      newMessage.innerHTML = `<strong>User:</strong> ${message}`;
      document.getElementById('chat').appendChild(newMessage);
    }

  </script>
</html>

💣 HTML structure for a single message is repeated in a few places. You can extract it into some JSX with Hono, but I didn't cover that here.

Want to Know More?

Stay tuned for more insights and tutorials! Visit My Blog 🤖

Creating Lucy: Developing an AI-Powered Slack Assistant with Memory

Kuba — Thu, 23 May 2024 10:01:22 +0000

Overview

The main idea was to build an AI agent named Lucy that can learn from our conversations and interact through a Slack interface. The plan was also to use no code tool (make.com) and play with embedding and pinecone.io. So the whole point was to build something that takes this:

And based on automatically creates table like this (I’m using airtable.com here).

Next I’m able to use those memories to create vectors in pinecone database and use them later in conversations.

💡 Be aware that the idea right now is to keep conversation’s history only until the memories are not distilled from it. Later we are operating only on memories, from this conversation which might lead to loosing some informations. But we can tackle this later…

Slack API

I won’t dive into the nitty-gritty details here, but the basic idea is simple. Create bot app and enable events api. We will use it to publish events on new messages in certain channel. Also I have turned on direct messages to the app. Then create make.com connection and voila.

Scenario - Lucy Slack conversation

📘 Check blueprint here

Let’s break it down step by step.

Custom Webhook - Start with a simple webhook to listen for events. Make sure to register this URL in your Slack app settings.
Webhook response - there is filter on route checking that event.type = url_verification. This is to confirm and send back challange to slack.
Next on router we check for event.type = message. Obviously we want to process only those events here.
Data store (Get a record) - fetching user slack id here. For later use.
Router - a few things here. Lets go one by one.
1. Webhook response - first send back webhook response. This to let slack know that we succeed in processing the event and to prevent resending this to us. I know that we don’t finish processing yet but that’s the mistake I can live with. I would rather prefer Lucy to not respond due to some error then responding multiple times and creating noise.
2. If the message if from the user - I’m just checking the author here. If its me → proceed. If the message comes from Lucy → Ignore.
  1. First fetch not synced conversation history. Aggregate it in format like:
```
kuba szwajka (2024-05-22T22:57:16.000Z): Hi! Whats my name? 
Lucy AI assistant (2024-05-22T22:57:16.000Z): Hey there! I don't have your name yet. What's your name? :smile: 
kuba szwajka (2024-05-22T22:57:26.000Z): my name is Kuba. Hi! 
Lucy AI assistant (2024-05-22T22:57:28.000Z): Hey Kuba! How's it going? :smile: 
```
  2. Then embed query sent to slack calling v1/embeddings endpoint with Make an API call. Next based on the resulting vector I’m querying my pinecone database for similar vectors (Assuming it will return vectors with informations that might be related to my query).
  3. More on those vectors and their metadata later but here I’m just taking vector [metadata.id](http://metadata.id) and based on that fetching memories database by ids.
  4. GPT Completion - here is the completion itself. The context at this place is build on two things. First - not synced conversation history that might contain useful facts. Second - useful memories that were found in pinecone.
  5. Create slack message
3. In the end, no matter who created message, I’m saving it with the author to ConversationHistory database and mark synced=false.

Scenario - Lucy find memories in conversation

📘 Check blueprint here

This scenario aims to process previously created conversation history, create memories and save them as a vectors. Right now it is triggered every 8 hours. Tbh. I’m testing if this is enough 🤷.

Nothing fancy here. Lets go step by step:

Fetch all records from ConversationHistory where synced=false.
Aggregate them as a text.
Let Chat distill the facts (facts overlapping to be implemented? Not sure if this will be the problem 🤔? Any thoughts? )
For the list of facts/memories, create a records in memories table with synced=false.
For each of these, create embedding calling v1/embeddings in OpenAI api.
Save it to pinecone and mark as synced.

Action

There are still a few missing pieces, like handling memory overlapping and preventing data loss during the conversation history to memory conversion.
I’ve put some blueprints if you want to try it out.

Want to Know More?

Stay tuned for more insights and tutorials! Visit My Blog 🤖

Creating Lucy: Your Daily Tasks delivered with ChatGPT and Todoist Integration

Kuba — Wed, 15 May 2024 09:00:42 +0000

Your Daily Tasks, Delivered: ChatGPT and Todoist Integration

Tools Used:

iOS Shortcuts: To trigger and manage high level automation.
Make.com: For automation across multiple services.
ChatGPT: To format messages.
Todoist: Task management.
Slack: To receive the task list via custom message. AI assistant interface.

Step-by-Step Guide

Configure Make.com automation:

In overall this automation looks like this:

First setup proper webhook as entrypoint for the automation.
Next we need to prepare the data we will use. Fetch Todoist tasks and parse the result to JSON. In Todoist block use "make api call" to /rest/v2/tasks and set Query String to filter=today | overdue. I’m using this to fetch only the list of tasks that are from all projects and date is overdue or today.
Send prepared data to ChatGPT. I’m using Chat Completion here.

Here is the System Message:
```
You are my daily task manager assistant. My name is Kuba. For the provided json list all tasks that are due today or overdue. Prepare a message informing me what is to be done today. 

current time: {{now}}

rules:
- you can use emoji.
- don't be too official
- adjust message to the time of a day
```
💡 Look at the current time property. There is a variable now from make.com to make chat aware what time we have. This way messages are more personalised to time of a day. In example sometimes it can suggest the most important task to tackle in the end of the day.

User Message is just a JSON string prepared in previous step.
Return the result. I’m using router here to return the result to iOS automation (for displaying notification on my phone) and to send it to the Slack. I think later I will keep slack only but right now its both.

Find blueprint here!

Configure iOS automation:

This is quite straightforward.

First call make.com webhook.
Use Show Result block to show Contents of URL from previous block.

Adjust automation on your phone to be triggered at specific hours. I’m using 9:00am and 6:00pm to have two checks automated.

Example results:

💡 Good morning, Kuba! ☕️ Here's what we've got on the agenda today:

1. Remind ... about the boat agreement 🚤
2. Update the training on Motivado 💡
3. Order a new wardrobe 🚪
4. Check the payment for the run 🏃‍♂️

Let's tackle these tasks to keep everything on track! Have a productive day!

Cheers, Your Assistant

💡 Hey Kuba! 🌟 Evenings are great for some reflection and maybe a bit of catching up, right? I see that you have a task that was due yesterday - you were supposed to "Uzupełnić trening Motivado." It seems like it's slipped by, but no worries, just consider squeezing it in tonight or setting aside time for it tomorrow. Keep up the good vibes! 💪🌙

As you can see it nicely formats longer lists and short ones (like one task). Also it adjusts the message to the time of a day! Another cool thing is that it understands multiple languages so my quick tasks I’m creating in polish language are also correctly translated and fit nicely in message.

Feel free to tweak and enhance the setup to fit your specific needs. If you have any questions, suggestions, or would like to share your experience, leave a comment below. Happy automating! 🚀

Want to Know More?

Stay tuned for more insights and tutorials! Visit My Blog 🤖