FastAPI Backend Development Bootcamp · บทเรียน

การรักษาความปลอดภัยระดับแถวและการแบ่งพาร์ทิชันข้อมูล

บังคับใช้ขอบเขตข้อมูลที่แยกจากกันอย่างเคร่งครัดด้วยการรักษาความปลอดภัยระดับแถวของ PostgreSQL และตัวป้องกันคำค้นหาตามขอบเขตผู้เช่า

บทเรียน 3 จาก 413 ขั้นตอน

การรักษาความปลอดภัยระดับแถวและการแบ่งพาร์ทิชันข้อมูล เป็นบทเรียน FastAPI Backend Development Bootcamp ฟรีบน CoddyKit นี่คือบทเรียนที่ 3 จากทั้งหมด 4 บทเรียน คุณสามารถอ่านบทเรียนทั้งหมดด้านล่างฟรี — จากนั้นลองปฏิบัติด้วยตัวคุณเองในเบราว์เซอร์พร้อมตัวแก้ไขโค้ดในตัวและติวเตอร์ AI ตลอด 24/7 บทเรียนนี้เป็นส่วนหนึ่งของเส้นทางการเรียน FastAPI Backend Development Bootcamp และความก้าวหน้าของคุณจะซิงค์ข้ามเว็บและแอป CoddyKit คอร์ส FastAPI Backend Development Bootcamp มีบทเรียนทั้งหมด 4 บทเรียน

บางส่วนของบทเรียนนี้ยังไม่ได้รับการแปล และแสดงเป็นภาษาอังกฤษ

Why Hard Tenant Boundaries Matter

In a multi-tenant SaaS, the most catastrophic bug class is cross-tenant data leakage: tenant A reading tenant B's rows. A single missing WHERE tenant_id = ? in one endpoint is enough to leak everything.

There are three layers of defense:

  • Application guards — every query scopes by tenant_id.
  • Database row-level security (RLS) — PostgreSQL refuses to return rows that don't match a session policy, even if the app forgets.
  • Partitioning — physically separate tenant data for performance and blast-radius control.

This lesson combines all three on FastAPI + PostgreSQL. The golden rule: never trust the application alone. RLS is the safety net that turns a leak into an empty result.

The Shared-Schema Tenant Model

We use the shared-schema model: one set of tables, every row carries a tenant_id column. It is the cheapest to operate and the easiest to migrate, but it has zero isolation by default — isolation is entirely your responsibility.

Every tenant-owned table follows the same shape: a tenant_id foreign key, indexed, and ideally part of a composite primary key so cross-tenant joins are impossible by construction.

from sqlalchemy import Column, BigInteger, String, ForeignKey, Index
from sqlalchemy.orm import declarative_base

Base = declarative_base()


class Invoice(Base):
    __tablename__ = "invoices"

    # Composite PK (tenant_id, id) makes a row globally addressable
    # only WITH its tenant -> no accidental cross-tenant lookups.
    tenant_id = Column(BigInteger, ForeignKey("tenants.id"), primary_key=True)
    id = Column(BigInteger, primary_key=True)
    customer = Column(String, nullable=False)
    amount_cents = Column(BigInteger, nullable=False)

    __table_args__ = (
        Index("ix_invoices_tenant", "tenant_id"),
    )

Resolving the Tenant From the Request

Every request must resolve to exactly one tenant before any query runs. Common sources: a subdomain (acme.app.com), a header (X-Tenant-ID), or — most securely — a claim baked into the authenticated JWT so the client cannot forge it.

Prefer the JWT claim. A header or subdomain is attacker-controlled; a signed token claim is not. Expose the resolved tenant through a FastAPI dependency so every endpoint shares one trusted source.

from fastapi import Depends, HTTPException, Request


async def get_current_tenant(request: Request) -> int:
    # Set earlier by the auth dependency after verifying the JWT signature.
    tenant_id = getattr(request.state, "tenant_id", None)
    if tenant_id is None:
        # Fail closed: no tenant context => refuse, never default to "all".
        raise HTTPException(status_code=401, detail="No tenant context")
    return tenant_id

Application-Level Query Guards

The first layer is disciplined query scoping. Centralize it so no developer has to remember the WHERE tenant_id = ? clause manually. A tenant-scoped session or repository injects the filter automatically.

Here a small repository wraps SQLAlchemy and forces the tenant filter on every read. The point is to make the safe path the default path.

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession


class InvoiceRepository:
    def __init__(self, session: AsyncSession, tenant_id: int):
        self.session = session
        self.tenant_id = tenant_id

    async def list(self):
        stmt = select(Invoice).where(Invoice.tenant_id == self.tenant_id)
        result = await self.session.execute(stmt)
        return result.scalars().all()

    async def get(self, invoice_id: int):
        stmt = select(Invoice).where(
            Invoice.tenant_id == self.tenant_id,
            Invoice.id == invoice_id,
        )
        return (await self.session.execute(stmt)).scalar_one_or_none()

Why App Guards Alone Are Not Enough

Application guards fail silently. A raw SQL report, a forgotten filter in an admin endpoint, an ORM relationship that loads across tenants, or a junior dev's hotfix — any of these bypasses the repository.

You want a layer that the application cannot forget. That is PostgreSQL Row-Level Security: the database itself attaches an invisible predicate to every query against a protected table. Even SELECT * FROM invoices returns only the current tenant's rows.

Enabling Row-Level Security in PostgreSQL

RLS works by binding policies to a session variable. The app sets app.current_tenant at the start of each request; policies compare each row's tenant_id against it.

Steps: enable RLS on the table, then create a USING policy (controls visibility for reads/updates/deletes) and a WITH CHECK policy (controls what inserts/updates are allowed to write).

-- Run once per protected table (migration)
ALTER TABLE invoices ENABLE ROW LEVEL SECURITY;
ALTER TABLE invoices FORCE ROW LEVEL SECURITY;  -- applies even to table owner

CREATE POLICY tenant_isolation ON invoices
    USING (tenant_id = current_setting('app.current_tenant')::bigint)
    WITH CHECK (tenant_id = current_setting('app.current_tenant')::bigint);

Setting the Tenant Context Per Request

For RLS to work, every connection must carry the right app.current_tenant before any query. With a connection pool this is delicate: a pooled connection is reused, so you must set the variable at request start and reset it at request end, otherwise a leaked variable causes cross-tenant reads.

Use set_config(key, value, true) — the true flag makes it transaction-local, so it auto-resets when the transaction ends. This is the safest pattern with pooled connections.

from sqlalchemy import text
from sqlalchemy.ext.asyncio import AsyncSession


async def scoped_session(session: AsyncSession, tenant_id: int):
    # Transaction-local: third arg `true` ties the setting to the current tx,
    # so it cannot leak to the next request reusing this pooled connection.
    await session.execute(
        text("SELECT set_config('app.current_tenant', :tid, true)"),
        {"tid": str(tenant_id)},
    )
    return session

Wiring RLS Into a FastAPI Dependency

Tie it together: a dependency yields a session that already has the tenant context set inside an open transaction. Endpoints then run normal queries and RLS quietly enforces isolation.

Note the connection must not run as a PostgreSQL superuser or table owner without FORCE ROW LEVEL SECURITY — superusers and owners bypass RLS by default. Use a dedicated, low-privilege application role.

from fastapi import Depends
from sqlalchemy.ext.asyncio import AsyncSession


async def get_tenant_session(
    tenant_id: int = Depends(get_current_tenant),
    session: AsyncSession = Depends(get_session),
):
    async with session.begin():  # one transaction => set_config stays scoped
        await session.execute(
            text("SELECT set_config('app.current_tenant', :tid, true)"),
            {"tid": str(tenant_id)},
        )
        yield session  # every query inside is RLS-filtered automatically

Partitioning by Tenant for Scale

Once you have many tenants, a single giant table hurts: vacuum, index bloat, and noisy-neighbor query plans. Declarative partitioning by tenant_id (LIST or HASH) splits the table into physical partitions.

Benefits: queries that include tenant_id get partition pruning (only the relevant partition is scanned), maintenance runs per-partition, and a tenant offboarding becomes a fast DETACH/DROP instead of a slow bulk delete.

-- Hash partitioning spreads tenants across N partitions evenly
CREATE TABLE invoices (
    tenant_id  bigint NOT NULL,
    id         bigint NOT NULL,
    customer   text   NOT NULL,
    amount_cents bigint NOT NULL,
    PRIMARY KEY (tenant_id, id)
) PARTITION BY HASH (tenant_id);

CREATE TABLE invoices_p0 PARTITION OF invoices
    FOR VALUES WITH (MODULUS 4, REMAINDER 0);
CREATE TABLE invoices_p1 PARTITION OF invoices
    FOR VALUES WITH (MODULUS 4, REMAINDER 1);
-- ... p2, p3

RLS and Partitioning Together

RLS policies on a partitioned parent table are inherited by all partitions automatically in modern PostgreSQL, so you define the policy once on the parent. Partition pruning and the RLS predicate compose: the planner prunes to one partition, then RLS filters rows within it.

Choosing LIST vs HASH:

  • HASH — even distribution, no hotspots, but you can't isolate one big tenant.
  • LIST — pin specific large tenants to dedicated partitions; great for noisy-neighbor isolation and per-tenant backups.

Verifying Isolation in Tests

Isolation must be tested like a security control, not assumed. Write a test that sets tenant A's context, inserts rows, switches to tenant B, and asserts B sees nothing. This pure-Python model demonstrates the exact invariant your RLS policy enforces.

class RLSSimulator:
    """Mimics how a USING policy filters rows by session tenant."""

    def __init__(self):
        self.rows = []
        self.current_tenant = None

    def set_tenant(self, tenant_id):
        self.current_tenant = tenant_id

    def insert(self, row):
        # WITH CHECK: can only write rows for the active tenant.
        if row["tenant_id"] != self.current_tenant:
            raise PermissionError("WITH CHECK violation")
        self.rows.append(row)

    def select_all(self):
        # USING: only rows matching the active tenant are visible.
        return [r for r in self.rows if r["tenant_id"] == self.current_tenant]


db = RLSSimulator()
db.set_tenant(1)
db.insert({"tenant_id": 1, "id": 10, "customer": "Acme"})

db.set_tenant(2)
db.insert({"tenant_id": 2, "id": 11, "customer": "Globex"})

print("Tenant 2 sees:", db.select_all())
db.set_tenant(1)
print("Tenant 1 sees:", db.select_all())

Quick Check: Pooled Connections and RLS

You set app.current_tenant with set_config on a connection drawn from a pool, but you used the session-level form (third argument false) instead of the transaction-local form. What is the most likely production consequence?

Recap: Defense in Depth for Tenant Data

You now have a layered isolation strategy for multi-tenant FastAPI:

  • App guards — a tenant-scoped repository makes the safe query the default; but it can be forgotten.
  • Row-Level Security — ENABLE + FORCE RLS, a USING/WITH CHECK policy keyed on app.current_tenant, and a low-privilege role so the DB cannot return foreign rows.
  • Transaction-local context — set the tenant with set_config(..., true) inside the request transaction so pooled connections never leak.
  • Partitioning — LIST/HASH by tenant_id for pruning, per-tenant maintenance, and fast offboarding; RLS policies inherit to all partitions.

The mindset: treat isolation as a security control with overlapping layers, and verify it with tests. Never trust the application alone.

เริ่มต้นได้ฟรี

เรียนรู้ FastAPI Backend Development Bootcamp ด้วย AI tutor — ฟรี

เขียนและเรียกใช้โค้ดจริงในเบราว์เซอร์ของคุณ รับความช่วยเหลือทันทีจาก AI tutor 24/7 และเรียนรู้ต่อจากที่คุณหยุดบนเว็บหรือในแอป

คอร์ส
21
บทเรียน
84

คำถามที่พบบ่อย

บทเรียน “การรักษาความปลอดภัยระดับแถวและการแบ่งพาร์ทิชันข้อมูล” ฟรีหรือไม่

ใช่ — ข้อความเต็มของ “การรักษาความปลอดภัยระดับแถวและการแบ่งพาร์ทิชันข้อมูล” ฟรีให้อ่านที่นี่บนเว็บ เพื่อปฏิบัติแบบโต้ตอบ (ตัวแก้ไขโค้ดในตัวและติวเตอร์ AI ตลอด 24/7) และปลดล็อคส่วนที่เหลือของคอร์ส FastAPI Backend Development Bootcamp ให้อัปเกรดเป็น CoddyKit PRO คอร์ส FastAPI Backend Development Bootcamp มีบทเรียนทั้งหมด 4 บทเรียน

คุณจะเรียนรู้อะไรในบทเรียน “การรักษาความปลอดภัยระดับแถวและการแบ่งพาร์ทิชันข้อมูล”

บังคับใช้ขอบเขตข้อมูลที่แยกจากกันอย่างเคร่งครัดด้วยการรักษาความปลอดภัยระดับแถวของ PostgreSQL และตัวป้องกันคำค้นหาตามขอบเขตผู้เช่า คุณปฏิบัติ FastAPI Backend Development Bootcamp ด้วยโค้ดที่ใช้งานได้จริงที่คุณเรียกใช้โดยตรงในเบราว์เซอร์ และติวเตอร์ AI ตลอด 24/7 ตอบคำถามของคุณขณะที่คุณไปผ่านบทเรียน

คุณต้องมีประสบการณ์ก่อนที่จะเริ่มเรียน FastAPI Backend Development Bootcamp หรือไม่

ไม่จำเป็นต้องมีประสบการณ์มาก่อน FastAPI Backend Development Bootcamp บน CoddyKit ออกแบบมาสำหรับผู้เริ่มต้นไปจนถึงผู้เรียนขั้นสูง คุณสามารถเริ่มต้นที่นี่หรือเริ่มจากตัวแรกและเรียนด้วยความเร็วของคุณเอง นี่คือบทเรียนที่ 3 จากทั้งหมด 4 บทเรียน

บทเรียน “การรักษาความปลอดภัยระดับแถวและการแบ่งพาร์ทิชันข้อมูล” ใช้เวลานานแค่ไหน

บทเรียน CoddyKit ส่วนใหญ่ใช้เวลาประมาณ 5–10 นาที แต่ละบทเรียนจึงสั้นและเป็นแบบโต้ตอบ คุณสามารถก้าวหน้าอย่างต่อเนื่องและกลับมาเรียนต่อจากตรงที่เพิ่งหยุดบนเว็บและแอปได้เลย

ฉันเขียนและรันโค้ดในบทเรียน FastAPI Backend Development Bootcamp นี้ได้ไหม

ได้ บทเรียน FastAPI Backend Development Bootcamp ทุกบทมีตัวแก้ไขโค้ดในตัว คุณจึงเขียนและรันโค้ดจริงได้เลยในเบราว์เซอร์ และได้รับข้อเสนอแนะจาก AI ในทันที — ไม่ต้องติดตั้งในเครื่องของคุณ

บทเรียนทั้งหมดในหลักสูตรนี้

  1. กลยุทธ์การแยกผู้เช่าและข้อแลกเปลี่ยน
  2. การระบุบริบทผู้เช่าและมิดเดิลแวร์
  3. การรักษาความปลอดภัยระดับแถวและการแบ่งพาร์ทิชันข้อมูล
  4. การวัดการใช้งาน โควตา และฮุกการเรียกเก็บเงิน
← กลับไปที่ FastAPI Backend Development Bootcamp