Archive

I’ve been using fastapi more and more lately and one feature I just started using is background tasks [[ thoughts-333 ]].

basic diskcache example <a href="/python-diskcache/" class="wikilink" data-title="How I setup a sqlite cache in python" data-description="When I need to cache some data between runs or share a cache accross multiple processes my go to library in python is . It&#39;s built on sqlite with just enough..." data-date="2022-03-29">How I setup a sqlite cache in python</a>

One Background Task per db entry #

I am using it for longer running tasks and I don’t want to give users the ability to spam these long running tasks with many duplicates running at the same time. And each fastapi worker will be running in a different process so I cannot keep track of work in memory, I have to do it in a distributed fashion. Since they are all running on the same machine with access to the same disk, diskcache is a good choice

What I need #

• check if a job is running
• automatically expire jobs

Less infrastructure complexity #

My brain first went to thinking I needed another service like redis running alongside fastapi for this, then it hit me that I can use diskcache.

How I used diskcache #

Here is how I used diskcache to debounce taking screenshots for a unique shot every 60 seconds.

from diskcache import Cache

jobs_cache = Cache("jobs-cache")

@shots_router.get("/shot/{shot_id}", responses={200: {"content": {"image/webp": {}}}})
@shots_router.get("/shot/{shot_id}/", responses={200: {"content": {"image/webp": {}}}})
async def get_shot_by_id(
    background_tasks: BackgroundTasks,
    request: Request,
    shot_id: int,
):
    shot = Shot.get(shot_id)
    # check if the shot exists and return it or continue to create it.



    is_running = jobs_cache.get(shot_id)

    if is_running:
        expire_time = datetime.fromtimestamp(jobs_cache.peekitem(expire_time=True)[1]) - datetime.now()
        console.print("[red]Already running store_shot: ", shot_id)
        console.print(f"[red]Can retry in {expire_time.seconds}s")
    else:
        jobs_cache.set(shot_id, True, 60)
        background_tasks.add_task(
            store_shot,
            shot_id=shot_id,
        )

Yesterday I realized that I have overlooked the default installation method of the sealed secrets controller for kubernetes kubeseal this whole time an jumped straight to the helm section. I spun up a quick kind cluster and had it up quickly. I can’t say this is any better or worse than helm as I have never needed to customize the install. According to the docs you can customize it with [[ kustomize ]] or helm.

# option if you don't have a cluster try with kind

kind create cluster

curl -L https://github.com/bitnami-labs/sealed-secrets/releases/download/v0.27.0/controller.yaml > controller.yaml

kubectl apply -f controller.yaml

I’ve long had issues with my qmk keyboard media keys on my arch install, I always thought it was on the keyboard end. Today I learned that playerctl fixes this.

paru -S playerctl

Once it is installed all of my media keys started working right away.

I played around with it a bit more and came up with a way to display the current playing title in my notifictations.

notify-send "`playerctl metadata --format '{{lc(status)}}:{{artist}}-{{album}}-{{title}}'`"

Today I am playing around with tailwind, flexing the css muscle and learning how to build new and different layouts with it.

I created a new post template that mimics a terminal look in css where I could inject the post title, description, and other frontmatter elements.