DEV Community: 郑沛沛

The Rise of Agentic AI: How LLMs Are Evolving Beyond Chatbots

郑沛沛 — Tue, 03 Mar 2026 07:16:38 +0000

The Rise of Agentic AI: How LLMs Are Evolving Beyond Chatbots

The landscape of artificial intelligence has undergone a seismic shift in recent years. Large Language Models (LLMs) have moved from simple text generators to sophisticated reasoning engines capable of autonomous decision-making. Welcome to the era of Agentic AI.

What is Agentic AI?

Agentic AI refers to systems that can perceive their environment, make decisions, and take actions independently to achieve specific goals. Unlike traditional chatbots that respond to prompts in isolation, agentic systems maintain state, plan multi-step strategies, and interact with external tools and APIs.

Key characteristics include:

Autonomy: Operating without constant human intervention
Goal-oriented behavior: Working toward defined objectives
Tool use: Leveraging external capabilities (calculators, databases, APIs)
Memory and context: Maintaining information across interactions
Planning and reasoning: Breaking complex tasks into manageable steps

The Evolution from GPT-3 to Modern Agents

When GPT-3 launched in 2020, it demonstrated impressive text generation but lacked true understanding. Today's models like GPT-4, Claude, and Gemini have evolved significantly:

Capability	GPT-3 Era	Modern Agents
Context Window	2K tokens	Up to 2M tokens
Tool Use	None	Native function calling
Reasoning	Pattern matching	Chain-of-thought reasoning
Memory	Stateless	Persistent vector stores
Multi-modal	Text only	Text, image, audio, video

Building Blocks of AI Agents

1. Planning and Reasoning

Modern agents use techniques like ReAct (Reasoning + Acting) and chain-of-thought prompting to break down complex problems:

# Example: ReAct pattern
thought = model.generate("What should I do to solve this task?")
action = parse_action(thought)
observation = execute_action(action)
# Loop until task complete

2. Memory Systems

Effective agents combine multiple memory types:

Short-term: Current conversation context
Long-term: Vector databases for semantic search
Episodic: Past interaction summaries
Procedural: Learned skills and workflows

3. Tool Integration

The most powerful agents don't work in isolation. They integrate with:

Search engines for real-time information
Code interpreters for computation
Databases for structured data access
APIs for external service interaction

Real-World Applications

Software Development

AI coding assistants like GitHub Copilot and Claude Code are evolving from autocomplete to full development partners:

Understanding entire codebases
Planning multi-file changes
Running tests and debugging
Managing dependencies and deployments

Research and Analysis

Agentic systems can:

Conduct literature reviews across thousands of papers
Synthesize findings into coherent reports
Identify gaps in existing research
Propose novel hypotheses

Business Operations

From customer service to supply chain management, agents are automating complex workflows:

Handling multi-turn support conversations
Processing invoices and purchase orders
Monitoring metrics and alerting stakeholders
Generating and scheduling content

Challenges and Considerations

Safety and Alignment

As agents gain autonomy, ensuring they act in accordance with human values becomes critical:

Reward hacking: Optimizing for wrong metrics
Power-seeking: Attempting to increase control
Deception: Misrepresenting capabilities or intentions

Reliability

Current agents still struggle with:

Consistency across long-running tasks
Recovery from errors gracefully
Understanding nuanced constraints
Avoiding hallucinations in critical contexts

Economic Impact

The rise of agentic AI raises important questions:

Job displacement vs. job transformation
Concentration of power among AI labs
Access inequality between organizations
Regulatory frameworks for autonomous systems

The Future Landscape

Looking ahead, we can expect several developments:

Multi-Agent Systems

Rather than single agents handling everything, specialized agents will collaborate:

Research agents gathering information
Analysis agents processing data
Writing agents generating content
Review agents checking quality

Embodied AI

Agents will increasingly interact with the physical world through robotics:

Warehouse automation
Healthcare assistance
Home maintenance robots
Autonomous vehicles

Personal AI Assistants

Everyone may have access to personalized agents that:

Manage schedules and communications
Handle routine administrative tasks
Provide tailored learning experiences
Act as creative collaborators

Getting Started with Agent Development

For developers interested in building agentic systems, here are key resources:

Frameworks

LangChain: Modular components for agent construction
AutoGPT: Experimental autonomous agent framework
Microsoft's AutoGen: Multi-agent conversation framework
CrewAI: Role-based agent collaboration

Best Practices

Start with clear, well-defined tasks
Implement robust error handling
Design for observability and debugging
Include human-in-the-loop checkpoints
Test extensively before deployment

Conclusion

Agentic AI represents a fundamental shift in how we interact with artificial intelligence. Moving beyond passive question-answering, these systems actively work to accomplish goals, making them powerful tools for productivity and innovation.

However, with great capability comes great responsibility. As we build more autonomous systems, we must prioritize safety, transparency, and alignment with human values. The future of AI isn't just about smarter models—it's about creating agents that genuinely serve human flourishing.

The age of digital colleagues has arrived. The question isn't whether AI agents will transform work and life, but how we'll shape that transformation for the benefit of all.

What applications of agentic AI are you most excited about? Share your thoughts in the comments below.

🚀 Level up your AI workflow! Check out my AI Developer Mega Prompt Pack — 80 battle-tested prompts for developers. $9.99

15 Coding Interview Questions I've Been Asked at Top Tech Companies

郑沛沛 — Mon, 02 Mar 2026 11:23:09 +0000

After interviewing at multiple tech companies, these are the questions that came up most often — with clean solutions and explanations.

1. Two Sum

The classic warm-up:

def two_sum(nums: list[int], target: int) -> list[int]:
    seen = {}
    for i, num in enumerate(nums):
        complement = target - num
        if complement in seen:
            return [seen[complement], i]
        seen[num] = i
    return []

# Time: O(n), Space: O(n)
assert two_sum([2, 7, 11, 15], 9) == [0, 1]

2. Valid Parentheses

def is_valid(s: str) -> bool:
    stack = []
    pairs = {')': '(', ']': '[', '}': '{'}
    for char in s:
        if char in '([{':
            stack.append(char)
        elif char in pairs:
            if not stack or stack[-1] != pairs[char]:
                return False
            stack.pop()
    return len(stack) == 0

assert is_valid("({[]})") == True
assert is_valid("([)]") == False

3. Merge Two Sorted Lists

class ListNode:
    def __init__(self, val=0, next=None):
        self.val = val
        self.next = next

def merge_two_lists(l1: ListNode, l2: ListNode) -> ListNode:
    dummy = ListNode()
    current = dummy
    while l1 and l2:
        if l1.val <= l2.val:
            current.next = l1
            l1 = l1.next
        else:
            current.next = l2
            l2 = l2.next
        current = current.next
    current.next = l1 or l2
    return dummy.next

4. Maximum Subarray (Kadane's Algorithm)

def max_subarray(nums: list[int]) -> int:
    max_sum = current_sum = nums[0]
    for num in nums[1:]:
        current_sum = max(num, current_sum + num)
        max_sum = max(max_sum, current_sum)
    return max_sum

assert max_subarray([-2, 1, -3, 4, -1, 2, 1, -5, 4]) == 6

5. LRU Cache

from collections import OrderedDict

class LRUCache:
    def __init__(self, capacity: int):
        self.cache = OrderedDict()
        self.capacity = capacity

    def get(self, key: int) -> int:
        if key not in self.cache:
            return -1
        self.cache.move_to_end(key)
        return self.cache[key]

    def put(self, key: int, value: int) -> None:
        if key in self.cache:
            self.cache.move_to_end(key)
        self.cache[key] = value
        if len(self.cache) > self.capacity:
            self.cache.popitem(last=False)

6. Binary Search

def binary_search(nums: list[int], target: int) -> int:
    left, right = 0, len(nums) - 1
    while left <= right:
        mid = (left + right) // 2
        if nums[mid] == target:
            return mid
        elif nums[mid] < target:
            left = mid + 1
        else:
            right = mid - 1
    return -1

7. Detect Cycle in Linked List

def has_cycle(head: ListNode) -> bool:
    slow = fast = head
    while fast and fast.next:
        slow = slow.next
        fast = fast.next.next
        if slow == fast:
            return True
    return False

8. Invert Binary Tree

def invert_tree(root):
    if not root:
        return None
    root.left, root.right = invert_tree(root.right), invert_tree(root.left)
    return root

9. Group Anagrams

from collections import defaultdict

def group_anagrams(strs: list[str]) -> list[list[str]]:
    groups = defaultdict(list)
    for s in strs:
        key = tuple(sorted(s))
        groups[key].append(s)
    return list(groups.values())

assert group_anagrams(["eat","tea","tan","ate","nat","bat"]) == [["eat","tea","ate"],["tan","nat"],["bat"]]

10. Top K Frequent Elements

from collections import Counter
import heapq

def top_k_frequent(nums: list[int], k: int) -> list[int]:
    count = Counter(nums)
    return heapq.nlargest(k, count.keys(), key=count.get)

assert top_k_frequent([1,1,1,2,2,3], 2) == [1, 2]

Interview Tips

Always clarify the problem before coding
Talk through your approach before writing code
Start with brute force, then optimize
Test with edge cases: empty input, single element, duplicates
Know your time and space complexity
Practice on LeetCode, but understand patterns, not just solutions

The 5 Patterns That Cover 80% of Questions

Two Pointers / Sliding Window
Hash Map for O(1) lookups
BFS/DFS for trees and graphs
Dynamic Programming (start with recursion + memoization)
Stack for matching/nesting problems

Key Takeaways

Most interviews test the same 20-30 patterns
Hash maps solve an surprising number of problems
Always consider time AND space complexity
Communication matters as much as the solution
Practice explaining your thought process out loud

6. It's okay to start with brute force and optimize

🚀 Level up your AI workflow! Check out my AI Developer Mega Prompt Pack — 80 battle-tested prompts for developers. $9.99

10 VS Code Extensions That Will Supercharge Your Development Workflow

郑沛沛 — Mon, 02 Mar 2026 11:22:22 +0000

VS Code is already great, but the right extensions make it exceptional. Here are 10 that I install on every machine, with configuration tips.

1. GitHub Copilot

AI pair programming that actually works:

// settings.json
{
    "github.copilot.enable": {
        "*": true,
        "markdown": true,
        "plaintext": false
    }
}

Pro tip: Write a descriptive comment before the function, and Copilot generates better code.

2. Error Lens

Shows errors and warnings inline, right next to the code:

{
    "errorLens.enabledDiagnosticLevels": ["error", "warning"],
    "errorLens.delay": 500
}

No more squinting at squiggly underlines — errors are impossible to miss.

3. GitLens

Git blame, history, and comparison built into the editor:

{
    "gitlens.codeLens.enabled": true,
    "gitlens.currentLine.enabled": true,
    "gitlens.hovers.currentLine.over": "line"
}

See who changed every line, when, and why — without leaving the editor.

4. REST Client

Test APIs directly from VS Code. Create .http files:

### Get all users
GET http://localhost:8000/api/users
Authorization: Bearer {{token}}

### Create user
POST http://localhost:8000/api/users
Content-Type: application/json

{
    "name": "Alice",
    "email": "alice@example.com"
}

### Variables
@token = eyJhbGciOiJIUzI1NiJ9...
@baseUrl = http://localhost:8000

Click "Send Request" above any block. Way faster than switching to Postman.

5. Thunder Client

If you prefer a GUI for API testing, Thunder Client is Postman inside VS Code. Collections, environments, and test scripts — all without leaving the editor.

6. Docker Extension

Manage containers, images, and compose files visually:

{
    "docker.containers.sortBy": "Status",
    "docker.showStartPage": false
}

Right-click to start, stop, view logs, or shell into any container.

7. Prettier + ESLint (for JS/TS projects)

{
    "editor.defaultFormatter": "esbenp.prettier-vscode",
    "editor.formatOnSave": true,
    "editor.codeActionsOnSave": {
        "source.fixAll.eslint": "explicit"
    },
    "[python]": {
        "editor.defaultFormatter": "charliermarsh.ruff"
    }
}

Format on save means you never think about formatting again.

8. Python + Ruff

Ruff replaces flake8, isort, and black — and it's 100x faster:

{
    "python.analysis.typeCheckingMode": "basic",
    "ruff.lint.run": "onSave",
    "ruff.format.args": ["--line-length", "100"]
}

9. Remote - SSH

Develop on remote servers as if they were local:

// ~/.ssh/config
Host dev-server
    HostName 10.0.0.50
    User developer
    IdentityFile ~/.ssh/id_ed25519

Open VS Code, connect to dev-server, and edit files with full IntelliSense. Game changer for cloud development.

10. Todo Tree

Find and organize TODOs across your codebase:

{
    "todo-tree.general.tags": ["TODO", "FIXME", "HACK", "BUG", "XXX"],
    "todo-tree.highlights.defaultHighlight": {
        "type": "text-and-comment"
    }
}

Never lose track of technical debt again.

Bonus: Essential Keyboard Shortcuts

Ctrl+P          Quick file open
Ctrl+Shift+P    Command palette
Ctrl+Shift+F    Search across files
Ctrl+D          Select next occurrence
Alt+Up/Down     Move line up/down
Ctrl+Shift+K    Delete line
Ctrl+`          Toggle terminal
Ctrl+B          Toggle sidebar
F2              Rename symbol
Ctrl+.          Quick fix

My settings.json Essentials

{
    "editor.fontSize": 14,
    "editor.lineHeight": 1.6,
    "editor.minimap.enabled": false,
    "editor.bracketPairColorization.enabled": true,
    "editor.guides.bracketPairs": true,
    "editor.stickyScroll.enabled": true,
    "files.autoSave": "onFocusChange",
    "terminal.integrated.fontSize": 13,
    "workbench.colorTheme": "One Dark Pro"
}

Key Takeaways

GitHub Copilot + good comments = faster coding
Error Lens makes errors impossible to miss
REST Client replaces Postman for most API testing
Format on save eliminates style debates
Remote SSH enables seamless cloud development

6. Learn keyboard shortcuts — they compound over time

🚀 Level up your AI workflow! Check out my AI Developer Mega Prompt Pack — 80 battle-tested prompts for developers. $9.99

Git Commands That Will Make You Look Like a Senior Developer

郑沛沛 — Mon, 02 Mar 2026 11:21:25 +0000

Most developers know add, commit, push. Here are the Git commands and workflows that separate juniors from seniors.

Interactive Rebase: Clean Up Your History

# Squash last 3 commits into one
git rebase -i HEAD~3

Your editor opens:

pick abc1234 Add user model
pick def5678 Fix typo in user model
pick ghi9012 Add validation to user model

Change to:

pick abc1234 Add user model
squash def5678 Fix typo in user model
squash ghi9012 Add validation to user model

Result: one clean commit instead of three messy ones.

Stash: Save Work Without Committing

# Save current changes
git stash

# Save with a description
git stash push -m "WIP: user authentication"

# List stashes
git stash list

# Apply most recent stash
git stash pop

# Apply specific stash
git stash apply stash@{2}

# Stash only specific files
git stash push -m "partial work" src/auth.py src/models.py

Bisect: Find the Bug-Introducing Commit

git bisect start
git bisect bad          # current commit is broken
git bisect good v1.0.0  # this tag was working

# Git checks out a middle commit. Test it, then:
git bisect good  # if this commit works
git bisect bad   # if this commit is broken

# Git narrows down to the exact commit that introduced the bug
# When done:
git bisect reset

Automate it:

git bisect start HEAD v1.0.0
git bisect run python -m pytest tests/test_auth.py

Cherry-Pick: Grab Specific Commits

# Apply a specific commit from another branch
git cherry-pick abc1234

# Cherry-pick without committing (stage changes only)
git cherry-pick --no-commit abc1234

# Cherry-pick a range
git cherry-pick abc1234..def5678

Reflog: Your Safety Net

# See everything you've done (even "deleted" commits)
git reflog

# Recover a "lost" commit after a bad reset
git reset --hard HEAD@{3}

# Recover a deleted branch
git checkout -b recovered-branch HEAD@{5}

Worktrees: Multiple Branches Simultaneously

# Work on a hotfix without switching branches
git worktree add ../hotfix-branch hotfix/critical-bug

# Now you have two directories, each on a different branch
cd ../hotfix-branch
# fix the bug, commit, push
cd ../main-repo

# Clean up
git worktree remove ../hotfix-branch

Better Diffs and Logs

# Word-level diff (great for prose/docs)
git diff --word-diff

# Show changes in a specific function
git log -p -L :function_name:file.py

# Pretty log with graph
git log --oneline --graph --all --decorate

# Find who changed a line
git blame -L 10,20 src/auth.py

# Search commit messages
git log --grep="fix auth"

# Search code changes
git log -S "def authenticate" --oneline

Aliases for Speed

# Add to ~/.gitconfig
git config --global alias.co checkout
git config --global alias.br branch
git config --global alias.st "status -sb"
git config --global alias.lg "log --oneline --graph --all --decorate"
git config --global alias.undo "reset --soft HEAD~1"
git config --global alias.amend "commit --amend --no-edit"

Commit Message Convention

# Format: type(scope): description
git commit -m "feat(auth): add JWT refresh token support"
git commit -m "fix(api): handle null response from payment gateway"
git commit -m "docs(readme): update installation instructions"
git commit -m "refactor(db): extract query builder into separate module"

# Types: feat, fix, docs, style, refactor, test, chore

Key Takeaways

Interactive rebase keeps your history clean and reviewable
Bisect finds bug-introducing commits automatically
Reflog is your safety net — almost nothing is truly lost in Git
Worktrees let you work on multiple branches without stashing
Good commit messages make debugging and reviewing much easier

6. Aliases save hundreds of keystrokes per day

🚀 Level up your AI workflow! Check out my AI Developer Mega Prompt Pack — 80 battle-tested prompts for developers. $9.99

PostgreSQL JSONB: When to Use It and How to Query It Like a Pro

郑沛沛 — Mon, 02 Mar 2026 11:19:34 +0000

PostgreSQL's JSONB gives you the flexibility of a document database with the reliability of SQL. Here's when and how to use it effectively.

When to Use JSONB

Good use cases:

User preferences and settings
API response caching
Event metadata with varying schemas
Feature flags and configuration

Bad use cases:

Core business data you query frequently
Data with strict relationships
Anything that needs foreign key constraints

Creating Tables with JSONB

CREATE TABLE events (
    id SERIAL PRIMARY KEY,
    event_type VARCHAR(50) NOT NULL,
    payload JSONB NOT NULL DEFAULT '{}',
    metadata JSONB,
    created_at TIMESTAMP DEFAULT NOW()
);

INSERT INTO events (event_type, payload, metadata) VALUES
('user.signup', '{"user_id": 123, "email": "alice@example.com", "plan": "pro"}', '{"ip": "192.168.1.1", "browser": "Chrome"}'),
('order.created', '{"order_id": 456, "items": [{"sku": "A1", "qty": 2}, {"sku": "B2", "qty": 1}], "total": 99.99}', '{"source": "web"}'),
('user.login', '{"user_id": 123, "method": "oauth", "provider": "google"}', '{"ip": "10.0.0.1"}');

Querying JSONB

Basic Access

-- Arrow operator (returns JSON)
SELECT payload->'email' FROM events WHERE event_type = 'user.signup';

-- Double arrow (returns text)
SELECT payload->>'email' FROM events WHERE event_type = 'user.signup';

-- Nested access
SELECT payload->'items'->0->>'sku' FROM events WHERE event_type = 'order.created';

Filtering

-- Filter by JSONB value
SELECT * FROM events WHERE payload->>'user_id' = '123';

-- Cast for numeric comparison
SELECT * FROM events WHERE (payload->>'total')::numeric > 50;

-- Check if key exists
SELECT * FROM events WHERE payload ? 'email';

-- Check if JSONB contains subset
SELECT * FROM events WHERE payload @> '{"plan": "pro"}';

Array Operations

-- Query inside JSON arrays
SELECT * FROM events
WHERE payload->'items' @> '[{"sku": "A1"}]';

-- Expand array elements
SELECT id, item
FROM events, jsonb_array_elements(payload->'items') AS item
WHERE event_type = 'order.created';

Indexing JSONB

-- GIN index: supports @>, ?, ?|, ?& operators
CREATE INDEX idx_events_payload ON events USING GIN(payload);

-- Specific path index (smaller, faster for known queries)
CREATE INDEX idx_events_user_id ON events ((payload->>'user_id'));
CREATE INDEX idx_events_plan ON events ((payload->>'plan'));

-- GIN with jsonb_path_ops (smaller index, only supports @>)
CREATE INDEX idx_events_payload_path ON events USING GIN(payload jsonb_path_ops);

JSONB Functions

-- Merge/update JSONB
UPDATE events
SET payload = payload || '{"verified": true}'
WHERE payload->>'user_id' = '123';

-- Remove a key
UPDATE events
SET payload = payload - 'temporary_field';

-- Set nested value
UPDATE events
SET payload = jsonb_set(payload, '{address,city}', '"New York"')
WHERE id = 1;

-- Pretty print
SELECT jsonb_pretty(payload) FROM events LIMIT 1;

-- Get all keys
SELECT DISTINCT jsonb_object_keys(payload) FROM events;

Python with SQLAlchemy

from sqlalchemy import Column, Integer, String, DateTime, func
from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy.orm import DeclarativeBase

class Base(DeclarativeBase):
    pass

class Event(Base):
    __tablename__ = "events"
    id = Column(Integer, primary_key=True)
    event_type = Column(String(50), nullable=False)
    payload = Column(JSONB, nullable=False, default={})
    created_at = Column(DateTime, server_default=func.now())

# Query JSONB fields
from sqlalchemy import cast, String

# Filter by JSONB value
events = session.query(Event).filter(
    Event.payload["user_id"].astext == "123"
).all()

# Contains operator
pro_signups = session.query(Event).filter(
    Event.payload.contains({"plan": "pro"})
).all()

# Insert with JSONB
new_event = Event(
    event_type="user.signup",
    payload={"user_id": 789, "email": "bob@example.com", "plan": "free"}
)
session.add(new_event)
session.commit()

Performance Tips

-- Check if your queries use indexes
EXPLAIN ANALYZE SELECT * FROM events WHERE payload @> '{"plan": "pro"}';

-- TOAST compression: large JSONB values are compressed automatically
-- But keep individual JSONB documents under 1MB for best performance

-- Use specific columns for frequently queried fields
ALTER TABLE events ADD COLUMN user_id INT GENERATED ALWAYS AS ((payload->>'user_id')::int) STORED;
CREATE INDEX idx_events_user_id_col ON events(user_id);

Key Takeaways

JSONB is great for semi-structured, varying data
Use GIN indexes for JSONB containment queries
Expression indexes for specific frequently-queried paths
@> (contains) is the most index-friendly operator
Use generated columns for frequently queried JSONB fields

6. Don't use JSONB for everything — structured data belongs in regular columns

🚀 Level up your AI workflow! Check out my AI Developer Mega Prompt Pack — 80 battle-tested prompts for developers. $9.99

Redis Beyond Caching: 7 Powerful Patterns You Should Know

郑沛沛 — Mon, 02 Mar 2026 11:17:47 +0000

Redis is more than a cache. It's a Swiss Army knife for distributed systems. Here are 7 patterns that solve real problems.

1. Rate Limiting

import redis
import time

r = redis.Redis()

def is_rate_limited(user_id: str, max_requests: int = 100, window: int = 60) -> bool:
    key = f"rate:{user_id}:{int(time.time()) // window}"
    current = r.incr(key)
    if current == 1:
        r.expire(key, window)
    return current > max_requests

# Usage
if is_rate_limited("user_123"):
    raise HTTPException(status_code=429, detail="Too many requests")

2. Distributed Locking

Prevent race conditions across multiple servers:

import uuid

def acquire_lock(lock_name: str, timeout: int = 10) -> str | None:
    lock_id = str(uuid.uuid4())
    if r.set(f"lock:{lock_name}", lock_id, nx=True, ex=timeout):
        return lock_id
    return None

def release_lock(lock_name: str, lock_id: str) -> bool:
    pipe = r.pipeline(True)
    key = f"lock:{lock_name}"
    try:
        pipe.watch(key)
        if pipe.get(key).decode() == lock_id:
            pipe.multi()
            pipe.delete(key)
            pipe.execute()
            return True
    except redis.WatchError:
        pass
    return False

# Usage
lock_id = acquire_lock("process_payment")
if lock_id:
    try:
        process_payment(order_id)
    finally:
        release_lock("process_payment", lock_id)

3. Leaderboards with Sorted Sets

# Add scores
r.zadd("leaderboard", {"alice": 2500, "bob": 1800, "charlie": 3200})

# Top 10 players
top_10 = r.zrevrange("leaderboard", 0, 9, withscores=True)
for rank, (player, score) in enumerate(top_10, 1):
    print(f"#{rank} {player.decode()}: {int(score)}")

# Player's rank
rank = r.zrevrank("leaderboard", "alice")  # 0-indexed

# Increment score
r.zincrby("leaderboard", 100, "alice")

4. Session Storage

import json

def create_session(user_id: str, data: dict, ttl: int = 3600) -> str:
    session_id = str(uuid.uuid4())
    r.setex(f"session:{session_id}", ttl, json.dumps({"user_id": user_id, **data}))
    return session_id

def get_session(session_id: str) -> dict | None:
    data = r.get(f"session:{session_id}")
    if data:
        r.expire(f"session:{session_id}", 3600)  # refresh TTL
        return json.loads(data)
    return None

def destroy_session(session_id: str):
    r.delete(f"session:{session_id}")

5. Pub/Sub for Real-Time Events

# Publisher
def publish_event(channel: str, event: dict):
    r.publish(channel, json.dumps(event))

publish_event("notifications", {"type": "new_order", "order_id": 123})

# Subscriber
def listen_events(channel: str):
    pubsub = r.pubsub()
    pubsub.subscribe(channel)
    for message in pubsub.listen():
        if message["type"] == "message":
            event = json.loads(message["data"])
            handle_event(event)

6. Queue with Reliability (BRPOPLPUSH)

def enqueue(queue: str, task: dict):
    r.lpush(queue, json.dumps(task))

def dequeue_reliable(queue: str, processing_queue: str, timeout: int = 0):
    data = r.brpoplpush(queue, processing_queue, timeout=timeout)
    if data:
        return json.loads(data)
    return None

def complete_task(processing_queue: str, task: dict):
    r.lrem(processing_queue, 1, json.dumps(task))

# Worker
while True:
    task = dequeue_reliable("tasks", "tasks:processing")
    if task:
        try:
            process(task)
            complete_task("tasks:processing", task)
        except Exception:
            # Task stays in processing queue for retry
            pass

7. Counting Unique Items with HyperLogLog

# Count unique visitors (uses only 12KB regardless of count!)
r.pfadd("visitors:2024-01-15", "user_1", "user_2", "user_3")
r.pfadd("visitors:2024-01-15", "user_1", "user_4")  # user_1 not counted again

unique_count = r.pfcount("visitors:2024-01-15")  # ~4

# Merge multiple days
r.pfmerge("visitors:week", "visitors:2024-01-15", "visitors:2024-01-16")
weekly_unique = r.pfcount("visitors:week")

Key Takeaways

Rate limiting with INCR + EXPIRE is simple and effective
Distributed locks prevent race conditions across servers
Sorted sets are perfect for leaderboards and rankings
Pub/Sub enables real-time event broadcasting
HyperLogLog counts unique items with minimal memory

6. Redis is a data structure server, not just a key-value store

🚀 Level up your AI workflow! Check out my AI Developer Mega Prompt Pack — 80 battle-tested prompts for developers. $9.99

Database Indexing Explained: How to Make Your Queries 1000x Faster

郑沛沛 — Mon, 02 Mar 2026 11:16:44 +0000

Slow queries are the #1 performance killer. Understanding indexes is the fastest way to fix them. Here's a practical guide.

What Is an Index?

An index is like a book's table of contents. Without it, the database scans every row (full table scan). With it, the database jumps directly to matching rows.

-- Without index: scans all 10 million rows
SELECT * FROM orders WHERE customer_id = 12345;
-- Time: 3200ms

-- With index: jumps to matching rows
CREATE INDEX idx_orders_customer ON orders(customer_id);
-- Same query now: 2ms

Types of Indexes

B-Tree (Default)

Best for equality and range queries:

CREATE INDEX idx_users_email ON users(email);
CREATE INDEX idx_orders_date ON orders(created_at);

-- These queries use the index:
SELECT * FROM users WHERE email = 'alice@example.com';
SELECT * FROM orders WHERE created_at > '2024-01-01';
SELECT * FROM orders WHERE created_at BETWEEN '2024-01-01' AND '2024-06-30';

Composite Indexes

Multiple columns — order matters:

CREATE INDEX idx_orders_status_date ON orders(status, created_at);

-- Uses the index (matches left-to-right):
SELECT * FROM orders WHERE status = 'shipped';
SELECT * FROM orders WHERE status = 'shipped' AND created_at > '2024-01-01';

-- Does NOT use the index efficiently:
SELECT * FROM orders WHERE created_at > '2024-01-01';  -- skips first column

Partial Indexes

Index only rows that matter:

-- Only index active users (much smaller index)
CREATE INDEX idx_active_users ON users(email) WHERE is_active = true;

-- Only index recent orders
CREATE INDEX idx_recent_orders ON orders(created_at) WHERE created_at > '2024-01-01';

GIN Index (for arrays and JSONB)

CREATE INDEX idx_tags ON posts USING GIN(tags);
SELECT * FROM posts WHERE tags @> ARRAY['python', 'tutorial'];

EXPLAIN ANALYZE: Your Best Friend

EXPLAIN ANALYZE SELECT * FROM orders WHERE customer_id = 12345;

-- Output:
-- Index Scan using idx_orders_customer on orders
--   Index Cond: (customer_id = 12345)
--   Planning Time: 0.1ms
--   Execution Time: 0.8ms

Watch for these red flags:

Seq Scan on large tables = missing index
Sort with high cost = consider index on sort column
Nested Loop with many rows = might need a join index

Common Anti-Patterns

-- BAD: Function on indexed column prevents index use
SELECT * FROM users WHERE LOWER(email) = 'alice@example.com';

-- GOOD: Use expression index
CREATE INDEX idx_users_email_lower ON users(LOWER(email));

-- BAD: Leading wildcard can't use index
SELECT * FROM users WHERE name LIKE '%alice%';

-- GOOD: Prefix search uses index
SELECT * FROM users WHERE name LIKE 'alice%';

-- BAD: OR can prevent index use
SELECT * FROM orders WHERE status = 'shipped' OR total > 1000;

-- GOOD: Use UNION
SELECT * FROM orders WHERE status = 'shipped'
UNION ALL
SELECT * FROM orders WHERE total > 1000 AND status != 'shipped';

Index Maintenance

-- Check index usage
SELECT indexrelname, idx_scan, idx_tup_read
FROM pg_stat_user_indexes
ORDER BY idx_scan ASC;

-- Find unused indexes (candidates for removal)
SELECT indexrelname, idx_scan
FROM pg_stat_user_indexes
WHERE idx_scan = 0 AND indexrelname NOT LIKE '%pkey%';

-- Check index size
SELECT pg_size_pretty(pg_indexes_size('orders'));

Python: Checking Query Performance

import time

async def profile_query(db, query, params=None):
    # Run EXPLAIN ANALYZE
    explain = await db.fetch_one(f"EXPLAIN ANALYZE {query}", params)
    print(explain)

    # Time the actual query
    start = time.perf_counter()
    result = await db.fetch_all(query, params)
    elapsed = time.perf_counter() - start
    print(f"Query returned {len(result)} rows in {elapsed:.3f}s")
    return result

Key Takeaways

Always index columns used in WHERE, JOIN, and ORDER BY
Composite index column order matters — most selective first
Use EXPLAIN ANALYZE to verify index usage
Partial indexes save space and improve performance
Remove unused indexes — they slow down writes

6. Functions on indexed columns prevent index use — use expression indexes

🚀 Level up your AI workflow! Check out my AI Developer Mega Prompt Pack — 80 battle-tested prompts for developers. $9.99

GraphQL with Python: Build Flexible APIs Using Strawberry

郑沛沛 — Mon, 02 Mar 2026 11:16:28 +0000

REST is great, but sometimes clients need more flexibility. GraphQL lets clients request exactly the data they need. Here's how to build a GraphQL API in Python.

Why GraphQL?

No over-fetching: client gets only requested fields
No under-fetching: get related data in one request
Strongly typed schema serves as documentation
Great for mobile apps and complex frontends

Setup with Strawberry

pip install strawberry-graphql[fastapi]

Basic Schema

import strawberry
from strawberry.fastapi import GraphQLRouter
from fastapi import FastAPI

@strawberry.type
class User:
    id: int
    name: str
    email: str

@strawberry.type
class Post:
    id: int
    title: str
    content: str
    author: User

# Sample data
users_db = [
    {"id": 1, "name": "Alice", "email": "alice@example.com"},
    {"id": 2, "name": "Bob", "email": "bob@example.com"},
]
posts_db = [
    {"id": 1, "title": "Hello GraphQL", "content": "GraphQL is awesome", "author_id": 1},
    {"id": 2, "title": "Python Tips", "content": "Use type hints", "author_id": 1},
    {"id": 3, "title": "Docker Guide", "content": "Containers 101", "author_id": 2},
]

@strawberry.type
class Query:
    @strawberry.field
    def users(self) -> list[User]:
        return [User(**u) for u in users_db]

    @strawberry.field
    def user(self, id: int) -> User | None:
        for u in users_db:
            if u["id"] == id:
                return User(**u)
        return None

    @strawberry.field
    def posts(self, limit: int = 10) -> list[Post]:
        results = []
        for p in posts_db[:limit]:
            author = next(u for u in users_db if u["id"] == p["author_id"])
            results.append(Post(id=p["id"], title=p["title"], content=p["content"], author=User(**author)))
        return results

schema = strawberry.Schema(query=Query)
app = FastAPI()
app.include_router(GraphQLRouter(schema), prefix="/graphql")

Querying

# Get only what you need
query {
  users {
    name
    email
  }
}

# Nested data in one request
query {
  posts(limit: 5) {
    title
    author {
      name
    }
  }
}

# Single user
query {
  user(id: 1) {
    name
    email
  }
}

Mutations

@strawberry.input
class CreatePostInput:
    title: str
    content: str
    author_id: int

@strawberry.type
class Mutation:
    @strawberry.mutation
    def create_post(self, input: CreatePostInput) -> Post:
        new_id = max(p["id"] for p in posts_db) + 1
        author = next(u for u in users_db if u["id"] == input.author_id)
        post_data = {
            "id": new_id,
            "title": input.title,
            "content": input.content,
            "author_id": input.author_id,
        }
        posts_db.append(post_data)
        return Post(id=new_id, title=input.title, content=input.content, author=User(**author))

    @strawberry.mutation
    def delete_post(self, id: int) -> bool:
        for i, p in enumerate(posts_db):
            if p["id"] == id:
                posts_db.pop(i)
                return True
        return False

schema = strawberry.Schema(query=Query, mutation=Mutation)

mutation {
  createPost(input: { title: "New Post", content: "Hello!", authorId: 1 }) {
    id
    title
    author { name }
  }
}

DataLoader: Solving N+1 Queries

from strawberry.dataloader import DataLoader

async def load_users(ids: list[int]) -> list[User]:
    # Single batch query instead of N queries
    users = await db.fetch_all("SELECT * FROM users WHERE id = ANY($1)", ids)
    user_map = {u["id"]: User(**u) for u in users}
    return [user_map.get(id) for id in ids]

user_loader = DataLoader(load_fn=load_users)

@strawberry.type
class Post:
    id: int
    title: str
    author_id: strawberry.Private[int]

    @strawberry.field
    async def author(self) -> User:
        return await user_loader.load(self.author_id)

Authentication in GraphQL

from strawberry.types import Info
from fastapi import Depends

async def get_context(current_user=Depends(get_current_user)):
    return {"user": current_user}

@strawberry.type
class Query:
    @strawberry.field
    def me(self, info: Info) -> User:
        user = info.context["user"]
        if not user:
            raise Exception("Not authenticated")
        return user

app.include_router(
    GraphQLRouter(schema, context_getter=get_context),
    prefix="/graphql"
)

Key Takeaways

GraphQL shines when clients need flexible data fetching
Strawberry provides a Pythonic, type-safe GraphQL experience
Use DataLoaders to prevent N+1 query problems
Mutations handle writes, queries handle reads
Authentication works through context injection

6. GraphQL playground at /graphql gives you interactive docs for free

🚀 Level up your AI workflow! Check out my AI Developer Mega Prompt Pack — 80 battle-tested prompts for developers. $9.99

Real-Time Apps with WebSockets in Python: Chat, Notifications, and Live Data

郑沛沛 — Mon, 02 Mar 2026 11:14:36 +0000

WebSockets enable real-time, bidirectional communication. Here's how to build real-time features that actually scale.

WebSocket vs HTTP

HTTP: Client asks, server answers. One request, one response.
WebSocket: Both sides can send messages anytime. Persistent connection.

Use WebSockets for: chat, live notifications, collaborative editing, live dashboards, multiplayer games.

Basic WebSocket Server with FastAPI

from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from typing import list

app = FastAPI()

class ConnectionManager:
    def __init__(self):
        self.active_connections: list[WebSocket] = []

    async def connect(self, websocket: WebSocket):
        await websocket.accept()
        self.active_connections.append(websocket)

    def disconnect(self, websocket: WebSocket):
        self.active_connections.remove(websocket)

    async def broadcast(self, message: str):
        for connection in self.active_connections:
            await connection.send_text(message)

    async def send_personal(self, message: str, websocket: WebSocket):
        await websocket.send_text(message)

manager = ConnectionManager()

@app.websocket("/ws/{username}")
async def websocket_endpoint(websocket: WebSocket, username: str):
    await manager.connect(websocket)
    await manager.broadcast(f"{username} joined the chat")
    try:
        while True:
            data = await websocket.receive_text()
            await manager.broadcast(f"{username}: {data}")
    except WebSocketDisconnect:
        manager.disconnect(websocket)
        await manager.broadcast(f"{username} left the chat")

JavaScript Client

const ws = new WebSocket('ws://localhost:8000/ws/alice');

ws.onopen = () => {
    console.log('Connected');
    ws.send('Hello everyone!');
};

ws.onmessage = (event) => {
    console.log('Received:', event.data);
    // Update UI
    const messages = document.getElementById('messages');
    messages.innerHTML += `<p>${event.data}</p>`;
};

ws.onclose = () => console.log('Disconnected');
ws.onerror = (error) => console.error('Error:', error);

Room-Based Chat

from collections import defaultdict

class RoomManager:
    def __init__(self):
        self.rooms: dict[str, list[WebSocket]] = defaultdict(list)

    async def join_room(self, room: str, websocket: WebSocket):
        await websocket.accept()
        self.rooms[room].append(websocket)

    def leave_room(self, room: str, websocket: WebSocket):
        self.rooms[room].remove(websocket)
        if not self.rooms[room]:
            del self.rooms[room]

    async def broadcast_to_room(self, room: str, message: str):
        for ws in self.rooms.get(room, []):
            await ws.send_text(message)

rooms = RoomManager()

@app.websocket("/ws/{room}/{username}")
async def room_chat(websocket: WebSocket, room: str, username: str):
    await rooms.join_room(room, websocket)
    await rooms.broadcast_to_room(room, f"{username} joined #{room}")
    try:
        while True:
            data = await websocket.receive_text()
            await rooms.broadcast_to_room(room, f"{username}: {data}")
    except WebSocketDisconnect:
        rooms.leave_room(room, websocket)
        await rooms.broadcast_to_room(room, f"{username} left #{room}")

JSON Messages with Types

import json
from pydantic import BaseModel

class WSMessage(BaseModel):
    type: str  # "chat", "typing", "status"
    data: dict

@app.websocket("/ws/{username}")
async def typed_websocket(websocket: WebSocket, username: str):
    await manager.connect(websocket)
    try:
        while True:
            raw = await websocket.receive_text()
            msg = WSMessage.model_validate_json(raw)

            if msg.type == "chat":
                await manager.broadcast(json.dumps({
                    "type": "chat",
                    "data": {"user": username, "text": msg.data["text"]}
                }))
            elif msg.type == "typing":
                await manager.broadcast(json.dumps({
                    "type": "typing",
                    "data": {"user": username}
                }))
    except WebSocketDisconnect:
        manager.disconnect(websocket)

Heartbeat / Keep-Alive

import asyncio

async def heartbeat(websocket: WebSocket):
    while True:
        try:
            await asyncio.sleep(30)
            await websocket.send_text(json.dumps({"type": "ping"}))
        except Exception:
            break

@app.websocket("/ws/{username}")
async def ws_with_heartbeat(websocket: WebSocket, username: str):
    await manager.connect(websocket)
    beat_task = asyncio.create_task(heartbeat(websocket))
    try:
        while True:
            data = await websocket.receive_text()
            msg = json.loads(data)
            if msg["type"] == "pong":
                continue
            await manager.broadcast(data)
    except WebSocketDisconnect:
        beat_task.cancel()
        manager.disconnect(websocket)

Key Takeaways

Use WebSockets for real-time bidirectional communication
ConnectionManager pattern handles multiple clients cleanly
Room-based architecture for scoped broadcasting
Use JSON messages with a type field for structured communication
Implement heartbeats to detect dead connections

6. For multi-server scaling, use Redis Pub/Sub as a message broker

🚀 Level up your AI workflow! Check out my AI Developer Mega Prompt Pack — 80 battle-tested prompts for developers. $9.99

Authentication Done Right: JWT, OAuth2, and Session Management

郑沛沛 — Mon, 02 Mar 2026 11:14:03 +0000

Authentication is the most security-critical part of your app. Here's how to implement it properly without reinventing the wheel.

JWT Authentication with FastAPI

from datetime import datetime, timedelta
from jose import JWTError, jwt
from passlib.context import CryptContext
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer

SECRET_KEY = "your-secret-key-from-env"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

def hash_password(password: str) -> str:
    return pwd_context.hash(password)

def verify_password(plain: str, hashed: str) -> bool:
    return pwd_context.verify(plain, hashed)

def create_access_token(data: dict, expires_delta: timedelta = None) -> str:
    to_encode = data.copy()
    expire = datetime.utcnow() + (expires_delta or timedelta(minutes=15))
    to_encode.update({"exp": expire})
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

async def get_current_user(token: str = Depends(oauth2_scheme)):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Invalid credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        user_id: str = payload.get("sub")
        if user_id is None:
            raise credentials_exception
    except JWTError:
        raise credentials_exception
    user = await get_user_by_id(user_id)
    if user is None:
        raise credentials_exception
    return user

Login Endpoint

from fastapi import APIRouter
from fastapi.security import OAuth2PasswordRequestForm

router = APIRouter()

@router.post("/token")
async def login(form: OAuth2PasswordRequestForm = Depends()):
    user = await authenticate_user(form.username, form.password)
    if not user:
        raise HTTPException(status_code=401, detail="Incorrect username or password")

    access_token = create_access_token(
        data={"sub": str(user.id)},
        expires_delta=timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    )
    return {"access_token": access_token, "token_type": "bearer"}

@router.get("/me")
async def read_users_me(current_user = Depends(get_current_user)):
    return current_user

Refresh Tokens

REFRESH_TOKEN_EXPIRE_DAYS = 7

def create_refresh_token(user_id: str) -> str:
    return create_access_token(
        data={"sub": user_id, "type": "refresh"},
        expires_delta=timedelta(days=REFRESH_TOKEN_EXPIRE_DAYS)
    )

@router.post("/refresh")
async def refresh_token(refresh_token: str):
    try:
        payload = jwt.decode(refresh_token, SECRET_KEY, algorithms=[ALGORITHM])
        if payload.get("type") != "refresh":
            raise HTTPException(status_code=401, detail="Invalid token type")
        user_id = payload.get("sub")
        new_access = create_access_token(data={"sub": user_id})
        return {"access_token": new_access, "token_type": "bearer"}
    except JWTError:
        raise HTTPException(status_code=401, detail="Invalid refresh token")

Role-Based Access Control

from enum import Enum
from functools import wraps

class Role(str, Enum):
    USER = "user"
    ADMIN = "admin"
    MODERATOR = "moderator"

def require_role(required_role: Role):
    def dependency(current_user = Depends(get_current_user)):
        if current_user.role != required_role:
            raise HTTPException(status_code=403, detail="Insufficient permissions")
        return current_user
    return dependency

@router.delete("/users/{user_id}")
async def delete_user(user_id: int, admin = Depends(require_role(Role.ADMIN))):
    await user_service.delete(user_id)
    return {"message": "User deleted"}

Rate Limiting

from slowapi import Limiter
from slowapi.util import get_remote_address

limiter = Limiter(key_func=get_remote_address)

@router.post("/login")
@limiter.limit("5/minute")
async def login(request: Request, form: OAuth2PasswordRequestForm = Depends()):
    # max 5 login attempts per minute per IP
    pass

Security Checklist

Hash passwords with bcrypt (never store plaintext)
Use short-lived access tokens (15-30 min)
Store refresh tokens securely (httpOnly cookies)
Rate limit authentication endpoints
Validate JWT signature AND expiration
Use HTTPS everywhere

Key Takeaways

JWT for stateless auth, sessions for server-side state
Always hash passwords with bcrypt or argon2
Implement refresh token rotation
Role-based access control with dependency injection
Rate limit login endpoints to prevent brute force

6. Never store secrets in code — use environment variables

🚀 Level up your AI workflow! Check out my AI Developer Mega Prompt Pack — 80 battle-tested prompts for developers. $9.99

FastAPI Beyond the Basics: Patterns for Production-Ready APIs

郑沛沛 — Mon, 02 Mar 2026 10:43:34 +0000

FastAPI is great for quick prototypes, but production apps need more structure. Here are patterns that scale.

Project Structure

myapp/
  app/
    __init__.py
    main.py
    config.py
    dependencies.py
    routers/
      __init__.py
      users.py
      orders.py
    models/
      __init__.py
      user.py
      order.py
    schemas/
      __init__.py
      user.py
      order.py
    services/
      __init__.py
      user_service.py

Configuration with Pydantic Settings

# app/config.py
from pydantic_settings import BaseSettings
from functools import lru_cache

class Settings(BaseSettings):
    database_url: str
    redis_url: str = "redis://localhost:6379"
    secret_key: str
    debug: bool = False
    allowed_origins: list[str] = ["http://localhost:3000"]

    class Config:
        env_file = ".env"

@lru_cache
def get_settings() -> Settings:
    return Settings()

Router Organization

# app/routers/users.py
from fastapi import APIRouter, Depends, HTTPException, status
from app.schemas.user import UserCreate, UserResponse
from app.services.user_service import UserService
from app.dependencies import get_user_service

router = APIRouter(prefix="/users", tags=["users"])

@router.post("/", response_model=UserResponse, status_code=status.HTTP_201_CREATED)
async def create_user(user: UserCreate, service: UserService = Depends(get_user_service)):
    existing = await service.get_by_email(user.email)
    if existing:
        raise HTTPException(status_code=409, detail="Email already registered")
    return await service.create(user)

@router.get("/{user_id}", response_model=UserResponse)
async def get_user(user_id: int, service: UserService = Depends(get_user_service)):
    user = await service.get(user_id)
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user

Dependency Injection

# app/dependencies.py
from fastapi import Depends
from sqlalchemy.ext.asyncio import AsyncSession
from app.database import get_db
from app.services.user_service import UserService

async def get_user_service(db: AsyncSession = Depends(get_db)) -> UserService:
    return UserService(db)

Error Handling

# app/main.py
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

app = FastAPI()

class AppException(Exception):
    def __init__(self, status_code: int, detail: str):
        self.status_code = status_code
        self.detail = detail

@app.exception_handler(AppException)
async def app_exception_handler(request: Request, exc: AppException):
    return JSONResponse(status_code=exc.status_code, content={"error": exc.detail})

@app.exception_handler(Exception)
async def generic_exception_handler(request: Request, exc: Exception):
    return JSONResponse(status_code=500, content={"error": "Internal server error"})

Middleware Stack

from fastapi.middleware.cors import CORSMiddleware
from fastapi.middleware.trustedhost import TrustedHostMiddleware
import time

app.add_middleware(
    CORSMiddleware,
    allow_origins=settings.allowed_origins,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

@app.middleware("http")
async def add_timing_header(request: Request, call_next):
    start = time.time()
    response = await call_next(request)
    response.headers["X-Process-Time"] = str(time.time() - start)
    return response

Background Tasks

from fastapi import BackgroundTasks

async def send_welcome_email(email: str):
    # slow email sending logic
    pass

@router.post("/register")
async def register(user: UserCreate, bg: BackgroundTasks):
    new_user = await service.create(user)
    bg.add_task(send_welcome_email, user.email)
    return new_user  # returns immediately

Key Takeaways

Separate routers, schemas, models, and services
Use Pydantic Settings for configuration
Dependency injection keeps code testable
Custom exception handlers for consistent error responses
Background tasks for non-blocking operations

6. Always add CORS middleware for frontend integration

🚀 Level up your AI workflow! Check out my AI Developer Mega Prompt Pack — 80 battle-tested prompts for developers. $9.99

Kubernetes for Beginners: Deploy Your First App in 15 Minutes

郑沛沛 — Mon, 02 Mar 2026 10:13:39 +0000

Kubernetes (K8s) orchestrates containers at scale. It sounds intimidating, but deploying your first app is surprisingly straightforward.

Core Concepts in 60 Seconds

Pod: Smallest unit — one or more containers running together
Deployment: Manages pods — handles scaling, updates, rollbacks
Service: Stable network endpoint for accessing pods
Namespace: Virtual cluster for organizing resources

Setup: Local K8s with minikube

# Install minikube
curl -LO https://storage.googleapis.com/minikube/releases/latest/minikube-linux-amd64
sudo install minikube-linux-amd64 /usr/local/bin/minikube

# Start cluster
minikube start

# Verify
kubectl get nodes

Deploy Your First App

Step 1: Create a Deployment

# deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: myapp
  labels:
    app: myapp
spec:
  replicas: 3
  selector:
    matchLabels:
      app: myapp
  template:
    metadata:
      labels:
        app: myapp
    spec:
      containers:
        - name: myapp
          image: myuser/myapp:v1
          ports:
            - containerPort: 8000
          resources:
            requests:
              memory: "128Mi"
              cpu: "100m"
            limits:
              memory: "256Mi"
              cpu: "500m"
          readinessProbe:
            httpGet:
              path: /health
              port: 8000
            initialDelaySeconds: 5
            periodSeconds: 10
          livenessProbe:
            httpGet:
              path: /health
              port: 8000
            initialDelaySeconds: 15
            periodSeconds: 20

kubectl apply -f deployment.yaml
kubectl get pods  # watch them come up

Step 2: Expose with a Service

# service.yaml
apiVersion: v1
kind: Service
metadata:
  name: myapp-service
spec:
  selector:
    app: myapp
  ports:
    - protocol: TCP
      port: 80
      targetPort: 8000
  type: LoadBalancer

kubectl apply -f service.yaml
minikube service myapp-service  # opens in browser

Step 3: ConfigMaps and Secrets

# configmap.yaml
apiVersion: v1
kind: ConfigMap
metadata:
  name: myapp-config
data:
  LOG_LEVEL: "info"
  MAX_CONNECTIONS: "100"

---
# secret.yaml
apiVersion: v1
kind: Secret
metadata:
  name: myapp-secrets
type: Opaque
stringData:
  DATABASE_URL: "postgresql://user:pass@db:5432/mydb"
  API_KEY: "sk-secret-key-here"

Use them in your deployment:

spec:
  containers:
    - name: myapp
      envFrom:
        - configMapRef:
            name: myapp-config
        - secretRef:
            name: myapp-secrets

Scaling

# Manual scaling
kubectl scale deployment myapp --replicas=5

# Auto-scaling
kubectl autoscale deployment myapp --min=2 --max=10 --cpu-percent=70

Rolling Updates

# Update image
kubectl set image deployment/myapp myapp=myuser/myapp:v2

# Watch rollout
kubectl rollout status deployment/myapp

# Rollback if something goes wrong
kubectl rollout undo deployment/myapp

Essential kubectl Commands

kubectl get pods                    # list pods
kubectl get pods -w                 # watch pods in real-time
kubectl describe pod <name>         # detailed pod info
kubectl logs <pod-name>             # view logs
kubectl logs <pod-name> -f          # stream logs
kubectl exec -it <pod-name> -- bash # shell into pod
kubectl top pods                    # resource usage
kubectl get events --sort-by='.lastTimestamp'  # recent events

Key Takeaways

Start with minikube for local development
Deployment + Service is the minimum viable K8s setup
Always set resource requests and limits
Use readiness and liveness probes for reliability
ConfigMaps for config, Secrets for sensitive data

6. Rolling updates give you zero-downtime deployments

🚀 Level up your AI workflow! Check out my AI Developer Mega Prompt Pack — 80 battle-tested prompts for developers. $9.99