Backend Engineering Assessment

Seven practical questions covering pagination, multi-tenancy, data integrity, Odoo ORM, PostgreSQL, FastAPI, and CI/CD.
Find what is wrong with each code snippet and select the correct answer.

0 / 18 answered

0 / 18

Medium

Q1 — Cursor-Based Pagination

You are building a REST API endpoint that allows an external sync service to pull a large list of records page by page. The endpoint uses cursor-based pagination instead of offset-based to avoid duplicate or skipped records when data changes between requests. The cursor encodes the position of the last record seen on the previous page. Records are always returned sorted by (updated_at ASC, id ASC).

import base64
from datetime import datetime
from typing import Optional


def encode_cursor(updated_at: datetime, record_id: int) -> str:
    dt_str = updated_at.isoformat() if updated_at else ""
    raw = f"{dt_str}|{record_id}"
    return base64.urlsafe_b64encode(raw.encode()).decode()


def decode_cursor(cursor: str):
    try:
        raw = base64.urlsafe_b64decode(cursor.encode()).decode()
        dt_str, id_str = raw.split("|")                     # line A
        updated_at = datetime.fromisoformat(dt_str) if dt_str else None
        return updated_at, int(id_str)
    except Exception as exc:
        raise ValueError("Invalid cursor") from exc


def build_cursor_filter(updated_at: Optional[datetime], record_id: int) -> dict:
    if updated_at:
        return {
            "updated_at__gte": updated_at,                  # line B
            "id__gt": record_id,                            # line C
        }
    return {"id__gt": record_id}


def get_page(db, table, base_filter, cursor, page_size):
    filters = dict(base_filter)
    if cursor:
        cur_date, cur_id = decode_cursor(cursor)
        filters.update(build_cursor_filter(cur_date, cur_id))

    rows = db.query(table).filter(**filters) \
             .order_by("updated_at", "id").limit(page_size).all()
    # line D                                      ^^^^^^^^^^

    has_more = len(rows) > page_size                        # line E
    rows = rows[:page_size]

    next_cursor = encode_cursor(rows[-1].updated_at, rows[-1].id) \
                  if (has_more and rows) else None
    return {"items": rows, "has_more": has_more, "next_cursor": next_cursor}

1.1 — Line A: raw.split("|") unpacks into exactly two variables. Describe a realistic input where this raises a ValueError. What is the correct fix and why?

A. Use raw.split("|", 1) — split from the left at most once B. Use raw.rsplit("|", 1) — split from the right at most once, so the ID is always the last segment C. Use raw.partition("|") — returns a 3-tuple which is safer D. The split is fine — base64 never contains |

rsplit("|", 1) is correct. The record ID is always the last segment. An ISO datetime can contain + characters but the raw format is datetime|id — splitting from the right ensures the ID is cleanly extracted even if the datetime portion ever contains a pipe. split("|", 1) splits from the left, which would put extra content in the wrong variable.

1.2 — Lines B & C: build_cursor_filter applies updated_at ≥ cursor_date AND id > cursor_id as a combined filter. Give a concrete example showing which records are silently dropped. What is the correct filter logic?

A. Should use updated_at__gt (strict greater than) instead of __gte B. Should use id__gte (greater than or equal) instead of id__gt C. The AND condition drops records where updated_at > cursor_date but id ≤ cursor_id — needs OR logic: (updated_at > cursor) OR (updated_at = cursor AND id > cursor_id) D. The filter should use updated_at__lte because cursors move backwards

Records are sorted (updated_at ASC, id ASC). A record with a newer timestamp but a lower ID (e.g. id=42, updated_at=T+1 after cursor id=100, updated_at=T) is silently dropped by the AND filter because 42 ≤ 100. The correct logic is an OR: either the timestamp is strictly greater, or the timestamp equals and the ID is strictly greater.

1.3 — Lines D & E: The query fetches exactly page_size rows, then checks len(rows) > page_size. What is has_more always equal to, and what is the effect on the caller?

A. has_more is always True — the database returns extra internal rows B. has_more is always False — the limit caps results at page_size, so the check never passes. Fix: use .limit(page_size + 1) C. has_more works correctly — .limit() is a hint, not a hard cap D. The issue is with next_cursor, not has_more

.limit(page_size) returns at most page_size rows. Therefore len(rows) > page_size is always False. The caller thinks there are no more pages after the first one. Fix: fetch page_size + 1 rows as a probe, check if the extra row exists, then trim.

Hard

Q2 — Tenant Data Isolation

You are working on a multi-tenant SaaS application. Each tenant is identified by a unique tenant_id. All data belongs to exactly one tenant and must never be visible to another. Each request is authenticated and the resolved tenant_id is passed down to the data layer. A helper function can_access_tenant(user, tenant_id) checks whether the authenticated user is permitted to work with the given tenant.

from typing import Optional


def can_access_tenant(user, tenant_id: int) -> bool:
    return tenant_id in user.allowed_tenant_ids


def build_contact_filter(
    role: str,
    contact_id: Optional[int] = None,
    modified_since: Optional[str] = None,
    external_id: Optional[str] = None,
) -> dict:
    filters = {
        "role": role,                       # line A
    }
    if contact_id:   filters["id"] = contact_id
    if modified_since: filters["updated_at__gte"] = modified_since
    if external_id:  filters["external_id"] = external_id
    return filters


def list_contacts(db, user, tenant_id, role, ...):
    if not can_access_tenant(user, tenant_id):      # line B
        raise PermissionError("Access denied")

    filters = build_contact_filter(role=role, ...)
    rows = db.query("contacts").filter(**filters).limit(page_size + 1).all()
    has_more = len(rows) > page_size
    rows = rows[:page_size]
    return {"items": rows, "has_more": has_more}

2.1 — can_access_tenant on line B is called before the query runs. A developer says "the access check is there, so tenants are isolated." Explain precisely why this is wrong.

A. can_access_tenant is called too late — it should run before parameter validation B. The access check verifies authorization (can this user work with this tenant?) but the query has no tenant_id filter — data scoping is completely missing, so rows from all tenants are returned C. The issue is that can_access_tenant should return the tenant object, not a boolean D. can_access_tenant doesn't validate the role parameter, so any role can be queried

Authorization ("is this user allowed?") and data scoping ("which rows belong to this tenant?") are two separate concerns. The access check passes, but the query runs against the entire contacts table with no tenant_id condition, returning records from all tenants.

2.2 — Where should the fix be applied?

A. Inside can_access_tenant — it should modify the query directly B. In list_contacts — add tenant_id to filters after calling build_contact_filter C. In build_contact_filter — require tenant_id as a parameter and always include it, so every caller inherits the fix D. In the database layer — use ORM row-level security as the sole protection

build_contact_filter is a shared query builder used by list_contacts, get_contact, update_contact, etc. Fixing it there means every caller automatically gets tenant scoping.

2.3 — Why is relying solely on ORM-level row security insufficient?

A. Privileged contexts (admin, background jobs) often bypass ORM rules, and a single protection layer is a single point of failure — an explicit filter provides defense in depth B. ORM row-level security is too slow for production workloads C. ORM rules cannot filter by tenant_id — they only work with user IDs D. ORM-level security only works with SQL databases, not NoSQL

ORM rules are often bypassed in privileged contexts (superuser queries, background jobs, migrations). An explicit tenant_id filter is always applied regardless of context, providing defense in depth.

Hard

Q3 — Atomic Write Operations

You are building an integration layer that receives contact records from an external sync service. When a contact is created, the system must perform two writes in sequence: (1) insert the contact into the local database, (2) write the external metadata (external_id, provider, synced_at). The external_id is the key used to deduplicate future sync calls. The sync service automatically retries any request that returns a 5xx response.

def find_existing_contact(db, tenant_id, external_id):
    return db.query("contacts").filter(
        tenant_id=tenant_id, external_id=external_id).first()


def create_contact(db, tenant_id, payload):
    external_id = payload.get("external_id")
    provider    = payload.get("provider")
    synced_at   = payload.get("synced_at")

    # Deduplicate: reject if this external_id already exists
    if external_id and find_existing_contact(db, tenant_id, external_id):
        raise ValueError(f"Contact already exists")

    # Step 1: insert the core contact record
    contact = db.insert("contacts", {
        "tenant_id": tenant_id,
        "name": payload.get("name"),
        "email": payload.get("email"),
    })
    # contact.id is now assigned. external_id is NOT yet written.

    # Step 2: write external metadata
    db.update("contacts", contact.id, {                     # line A
        "external_id": external_id,
        "provider": provider,
        "synced_at": synced_at,
    })
    return db.query("contacts").filter(id=contact.id).first()

3.1 — Trace this failure: (1) sync sends external_id="CRM-99", (2) db.insert succeeds (id=200), (3) db.update on line A raises DatabaseError, (4) HTTP 500 returned, (5) sync retries. What does find_existing_contact return in step 5? What happens next?

A. The retry finds the existing contact via external_id and raises ValueError — deduplication works correctly B. The retry fails with the same DatabaseError — the record is locked C. find_existing_contact returns None because external_id was never written — a duplicate orphan row is created D. The database automatically rolls back the insert when the update fails

The insert committed with external_id=NULL. The update failed, so the metadata was never written. On retry, find_existing_contact searches for external_id="CRM-99" and finds nothing. A second row is created, producing a duplicate orphan.

3.2 — What is the correct database primitive to make both writes atomic?

A. A table lock on the contacts table during the two writes B. A savepoint (nested transaction) wrapping both insert and update — if the update fails, the insert is rolled back too C. An optimistic concurrency check using a version column D. A unique constraint on external_id — the database rejects duplicates automatically

A savepoint groups the insert and update so both succeed or both roll back. If the update raises, the savepoint undoes the insert. The database returns to the state before create_contact was called.

3.3 — A colleague suggests try/except with a manual delete if the update fails. Identify two problems with this approach.

A. The delete itself can fail (same DB issue), AND concurrent queries can read the orphan row between insert and delete (visibility window) B. The delete is too slow for high-throughput systems, AND it creates excessive WAL entries C. The delete requires a separate transaction, AND it violates foreign key constraints D. The delete bypasses audit logging, AND it requires elevated database permissions

Problem 1: If the DB raised on update, db.delete may fail for the same reason, leaving the orphan. Problem 2: Between insert and delete, the orphan row is visible to concurrent sessions. A savepoint keeps the insert invisible until the full block commits.

Hard

Q4 — Odoo ORM / Model Inheritance

An Odoo 17 module defines a model to track project budgets. The developer extended the existing project.project model with computed fields and a record rule for company-level isolation. Review the code for correctness.

from odoo import models, fields, api


class ProjectBudget(models.Model):
    _name = 'project.budget'        # line A
    _inherit = 'project.project'    # line B

    budget = fields.Float(string='Budget')
    spent = fields.Float(string='Spent')
    remaining = fields.Float(
        string='Remaining',
        compute='_compute_remaining',
        store=True,
    )

    @api.onchange('budget')         # line C
    def _compute_remaining(self):
        for rec in self:
            rec.remaining = rec.budget - rec.spent

<!-- Record rule XML -->
<record model="ir.rule" id="rule_budget_own_company">
  <field name="name">Budget: own company</field>
  <field name="model_id" ref="model_project_budget"/>
  <field name="domain_force">
    [('company_id','=',user.company_id.id)]
  </field>
  <field name="groups" eval="[]"/>           <!-- line D -->
</record>

4.1 — Lines A & B: The developer set both _name = 'project.budget' and _inherit = 'project.project'. What does this produce in Odoo?

A. It extends project.project in-place, adding the new fields to the existing table — this is correct for adding budget tracking B. It creates a brand new model project.budget with its own database table that copies all fields from project.project (prototype inheritance) — the developer probably wanted _inherit without redefining _name C. It raises a ValueError at module install because _name and _inherit cannot be different D. It creates an alias — both names point to the same table and can be used interchangeably

In Odoo, when _name differs from _inherit, Odoo creates a brand new model with its own table (prototype inheritance). The new model copies all fields from the parent but stores data separately. Existing project.project records do NOT gain the budget fields. To extend in-place, use _inherit = 'project.project' without redefining _name.

4.2 — Line C: The method _compute_remaining uses @api.onchange('budget') but the field declaration says compute='_compute_remaining', store=True. What breaks?

A. Nothing — @api.onchange and compute= work the same way for stored fields B. The field will only recompute during UI form interactions, NOT on ORM write() or create() calls, background jobs, or XML-RPC. Missing spent dependency. Should use @api.depends('budget', 'spent') C. store=True is incompatible with @api.onchange and will raise a ValidationError D. @api.onchange and @api.depends are identical — the only issue is that spent is missing from the decorator arguments

@api.onchange only triggers during form interactions in the web client. It does NOT trigger on direct ORM write()/create() calls, background jobs, or XML-RPC. A stored computed field must use @api.depends('budget', 'spent') so the ORM recomputes whenever either dependency changes.

4.3 — Line D: The record rule sets groups to an empty list eval="[]". What is the security consequence?

A. The rule applies to no groups, so it has no effect — budget records are unrestricted B. The rule applies to the base.group_user group only, which is correct for internal users C. The rule becomes a global rule for ALL users — this is usually intentional for company-level isolation and is correct here D. The rule becomes a global rule enforced for ALL users including sudo() — records with company_id = NULL become invisible to everyone, including administrators

In Odoo, a record rule with empty groups is a global rule. Unlike group-specific rules (additive, bypassed by superuser), global rules are enforced for ALL users and cannot be bypassed even with sudo(). If any project.budget record has company_id = False, it becomes completely invisible to everyone.

Medium

Q5 — PostgreSQL Query & Migration

A backend team maintains a transactions table with ~8 million rows. They need to optimize a slow query and apply a schema migration. Review both operations.

-- Table definition:
CREATE TABLE transactions (
    id          BIGSERIAL PRIMARY KEY,
    tenant_id   UUID NOT NULL,
    status      VARCHAR(20) NOT NULL DEFAULT 'pending',
    amount      NUMERIC(12,2) NOT NULL,
    created_at  TIMESTAMPTZ NOT NULL DEFAULT now()
);

-- Existing index:
CREATE INDEX idx_transactions_tenant
    ON transactions (tenant_id, created_at);

-- Slow query (takes 4.2s):
SELECT id, amount, created_at
FROM transactions
WHERE tenant_id = '...'
  AND status = 'completed'                    -- line A
  AND created_at >= '2025-01-01'
ORDER BY created_at DESC
LIMIT 50;

-- EXPLAIN ANALYZE output (excerpt):
--  Sort  (cost=28451..28452 rows=50)
--    -> Filter  (rows=50, rows removed by filter: 184720)  -- line B
--         -> Index Scan using idx_transactions_tenant
--             (rows=184770)

-- Migration script:
ALTER TABLE transactions
    ADD COLUMN notes TEXT NOT NULL;            -- line C

5.1 — Lines A & B: The EXPLAIN output shows the index scan returns 184,770 rows but the filter removes 184,720, leaving only 50. What is the root cause and the fix?

A. The query is missing LIMIT — adding it would stop the scan early B. The index (tenant_id, created_at) does not cover the status column, so PostgreSQL fetches all matching rows and filters status = 'completed' in memory. Fix: composite index (tenant_id, status, created_at) C. The status column needs to be changed from VARCHAR to an ENUM type for faster comparison D. The created_at >= '2025-01-01' clause uses a string instead of a timestamp, causing a type cast that prevents index usage

The existing index (tenant_id, created_at) lets PostgreSQL find rows for a given tenant within a date range, but it knows nothing about status. The database fetches all 184,770 matching rows, checks status = 'completed' on each, and discards 99.97%. A composite index (tenant_id, status, created_at) allows seeking directly to the right status.

5.2 — Line C: ALTER TABLE transactions ADD COLUMN notes TEXT NOT NULL on an 8-million-row table. What is the operational problem?

A. TEXT columns cannot have a NOT NULL constraint — PostgreSQL will reject the statement B. Adding a NOT NULL column without a DEFAULT fails (existing rows have NULL) or requires a table rewrite with ACCESS EXCLUSIVE lock. Fix: ADD COLUMN notes TEXT NOT NULL DEFAULT '' — PostgreSQL 11+ writes the default to metadata only, avoiding a table rewrite C. Existing rows will have NULL in the notes column, violating the constraint on the next read D. PostgreSQL handles schema changes online without locks — the NOT NULL constraint is fine

When you add a NOT NULL column without a DEFAULT, PostgreSQL fails with ERROR: column contains null values. With a DEFAULT on PG 11+, the value is written to catalog metadata without touching rows — nearly instant and no lock.

Medium

Q6 — FastAPI + Cloud Function

A Google Cloud Function (Python, HTTP-triggered) wraps a FastAPI app. It calls an external payment API and returns the result. Review the code for correctness and security.

import functions_framework
import httpx, os
from fastapi import FastAPI, Request
from mangum import Mangum

app = FastAPI()

@app.post("/charge")
async def charge(request: Request):
    body = await request.json()
    amount = body["amount"]
    token = body["payment_token"]
    api_key = os.environ.get("PAYMENT_API_KEY")

    async with httpx.AsyncClient() as client:
        resp = await client.post(
            "https://api.payments.example.com/v1/charges",
            json={"amount": amount, "source": token},
            headers={"Authorization": f"Bearer {api_key}"},
            timeout=30,
        )

    if resp.status_code != 200:
        return {                                          # line B
            "error": True,
            "detail": resp.text,
            "headers": dict(resp.headers),
        }
    return {"success": True, "charge_id": resp.json()["id"]}


@app.post("/refund")
def refund(request: Request):                             # line C
    body = request.json()                                 # line C (cont.)
    # ... process refund ...
    return {"success": True}

@functions_framework.http
def main(request):
    handler = Mangum(app)
    return handler(request)

6.1 — Line B: When the payment API returns an error, the code returns resp.text and resp.headers to the caller. What are the two problems?

A. resp.text might be empty, and resp.headers cannot be serialized to JSON B. Leaks internal details from the upstream API (error messages, headers, rate-limit info) to the client, AND returns HTTP 200 with "error": True instead of a proper error status code (502 Bad Gateway) C. Performance — constructing the headers dict is slow for large responses D. resp.text needs to be decoded from bytes first, and dict(resp.headers) drops duplicate header keys

Returning raw upstream error text and headers to the client is an information leak. The upstream API's errors may contain internal references and infrastructure details. Additionally, the function returns HTTP 200 with {"error": True} — clients checking HTTP status codes think the request succeeded. Should return 502 with a generic error message.

6.2 — Line C: The /refund endpoint is def refund() (synchronous, no async) and calls request.json() without await. What happens at runtime?

A. It works correctly — FastAPI automatically wraps synchronous handlers in a thread pool B. request.json() returns a coroutine object instead of the parsed body. Accessing body["amount"] raises TypeError: 'coroutine' object is not subscriptable. Fix: async def + await C. The function runs correctly but blocks the event loop — the fix is to add async but await is not needed D. FastAPI rejects synchronous handlers entirely and raises a startup error

In FastAPI/Starlette, Request.json() is an async method that returns a coroutine. Calling it without await returns the coroutine object itself, not the parsed JSON. Any dictionary access on this coroutine raises TypeError. Fix: async def refund() and body = await request.json().

Medium

Q7 — CI/CD: GitHub Actions + Cloud Run

A team deploys a multi-service application. The CI/CD pipeline runs checks on PRs and auto-deploys on merge to main. The deployment workflow deploys interdependent Cloud Run services in sequence. Review both workflows.

# deploy.yml
name: Deploy to Cloud Run
on:
  push:
    branches: [main]

jobs:
  deploy:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - id: auth
        uses: google-github-actions/auth@v2
        with:
          workload_identity_provider: ${{ secrets.WIF_PROVIDER }}
          service_account: ${{ secrets.WIF_SERVICE_ACCOUNT }}

      - name: Deploy API service
        run: |
          gcloud run deploy api-service \
            --image=europe-west1-docker.pkg.dev/proj/repo/api:${{ github.sha }} \
            --region=europe-west1

      - name: Deploy worker service
        run: |
          WORKER_URL=$(gcloud run services describe worker-service \  # line B
            --region=europe-west1 --format='value(status.url)')
          gcloud run deploy worker-service \
            --image=europe-west1-docker.pkg.dev/proj/repo/worker:${{ github.sha }} \
            --region=europe-west1 \
            --set-env-vars=API_URL=https://api-service-xxxxx.run.app  # line C

# pr-check.yml
name: PR Checks
on:
  pull_request:
    branches: [main]

jobs:
  deploy-preview:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - id: auth
        uses: google-github-actions/auth@v2
        with:
          credentials_json: ${{ secrets.GCP_SA_KEY }}        # line D
      - name: Deploy preview
        run: |
          gcloud run deploy preview-${{ github.event.pull_request.number }} \
            --image=europe-west1-docker.pkg.dev/proj/repo/api:${{ github.sha }} \
            --region=europe-west1

7.1 — Lines B & C: The worker deployment reads the CURRENT worker URL (line B) but hardcodes the API URL (line C). What are the two problems?

A. gcloud run services describe is slow — should cache the URL B. Line B reads the OLD worker URL before deploying (useless — Cloud Run URLs are stable per service). Line C hardcodes api-service-xxxxx.run.app — if the API service URL ever changes, the worker silently connects to a stale endpoint. Fix: read API URL dynamically after deploying it C. The worker should be deployed before the API service to avoid downtime D. Cloud Run URLs are deterministic — hardcoding is fine since they never change

Line B reads the worker URL before deploying the new version (pointless — the URL doesn't change). Line C hardcodes a URL hash that will break if the API service is recreated, moved to a different region, or renamed. The fix is to read the API URL dynamically: API_URL=$(gcloud run services describe api-service --format='value(status.url)').

7.2 — Line D: The PR check workflow uses credentials_json (a service account JSON key). The deploy workflow uses Workload Identity Federation. What is the security problem?

A. JSON keys are slower than WIF — performance issue only B. SA JSON keys are long-lived credentials that never expire — if the GitHub secret is leaked, the attacker has permanent GCP access until the key is manually rotated. WIF uses short-lived tokens scoped to the workflow run C. JSON keys don't support multi-project access D. credentials_json requires base64 encoding — the workflow will fail at auth

SA JSON keys are long-lived credentials that never expire. If the GitHub secret is leaked (repo compromise, log exposure, fork access), the attacker has permanent GCP access until the key is manually revoked. Workload Identity Federation uses short-lived tokens scoped to the specific workflow run, with no persistent credentials to leak.