codesnips
python 
from sqlalchemy import select
from sqlalchemy.orm import Session

from .models import User

BATCH_SIZE = 1000

Stream a Large CSV Export in FastAPI With StreamingResponse and a Generator

fastapi streaming csv
by codesnips 3 tabs
python 
from dataclasses import dataclass, field
from datetime import datetime, timezone
from decimal import Decimal


def _now():

Synchronous Domain Event Bus with a Registry-Based Publisher in Python

domain-events event-bus pubsub
by codesnips 3 tabs
python 
import base64
import json
from datetime import datetime


class InvalidCursor(Exception):

Cursor-Based Pagination for a Flask JSON API Blueprint

flask pagination cursor
by codesnips 3 tabs
ruby 
class CreateAuditLogs < ActiveRecord::Migration[7.1]
  def change
    create_table :audit_logs do |t|
      t.references :auditable, polymorphic: true, null: false
      t.references :user, null: true, foreign_key: true
      t.string :action, null: false

Rails Audit Log for Model Changes Using ActiveRecord Callbacks

rails activerecord callbacks
by codesnips 4 tabs
rust 
use std::time::Instant;

use axum::{
    extract::Request,
    http::HeaderValue,
    middleware::Next,

Axum Latency Logging Middleware With Request IDs and Tracing Spans

axum tower middleware
by codesnips 2 tabs
java 
package com.example.api.error;

import com.fasterxml.jackson.annotation.JsonInclude;

import java.time.Instant;
import java.util.List;

Consistent JSON Error Responses with @RestControllerAdvice in Spring Boot

spring-boot rest-api error-handling
by codesnips 4 tabs
ruby 
module Notifications
  class RecipientSerializer < ActiveJob::Serializers::ObjectSerializer
    def serialize?(argument)
      argument.is_a?(Notifications::Recipient)
    end

Enqueuing Notification Batches with a Custom Active Job Argument Serializer for Value Objects

rails active-job serialization
by codesnips 4 tabs
ruby 
require "rack/attack"
require "redis-store"

class Rack::Attack
  cache.store = Redis::Store.new(url: ENV.fetch("REDIS_URL", "redis://localhost:6379/1"))

Rate-Limiting a Sinatra JSON API With Rack::Attack Throttles and Redis

sinatra rack rack-attack
by codesnips 3 tabs
python 
import errno
import fcntl
import os
import time

File-Based Locking to Prevent Concurrent Cron Job Runs in Python

locking concurrency fcntl
by codesnips 3 tabs
rust 
use std::any::Any;

pub trait CommandHandler {
    fn name(&self) -> &str;

    fn handle(&self, input: &dyn Any) -> Box<dyn Any>;

Type-Erased Command Handler Registry With Trait Objects in Rust

rust trait-objects dispatch
by codesnips 3 tabs
python 
import functools
import threading


def debounce(wait):
    def decorator(func):

Debounce Repeated Function Calls With a Thread-Safe Python Decorator

python decorators debounce
by codesnips 2 tabs
typescript 
import { SetMetadata } from '@nestjs/common';

export enum Role {
  User = 'user',
  Editor = 'editor',
  Admin = 'admin',

Role-Based Access Control in NestJS with a Custom Guard and @Roles Decorator

nestjs authorization rbac
by codesnips 3 tabs