Backend (Flask)

The NexSchool backend is a Flask app built with the application-factory pattern: one factory wires config, extensions, ~30 blueprints, tenant middleware, and error handlers, with per-module routes / services / models / schemas and multi-tenant query scoping enforced in the core layer.

App factory & blueprints

server/app.py exposes create_app(config_name=None). It builds the Flask app, wraps it in ProxyFix (so request.remote_addr is the real client IP behind nginx — required for rate-limiting, login-lockout keying, and audit logs), loads a config class, then initializes everything in order.

def create_app(config_name=None):
    app = Flask(__name__)
 
    # Behind nginx (one proxy hop): trust X-Forwarded-* so request.remote_addr
    # is the real client IP — else rate-limiting/lockout/audit see nginx's IP.
    app.wsgi_app = ProxyFix(app.wsgi_app, x_for=1, x_proto=1, x_host=1)
 
    config_class = get_config(config_name)   # dev / production / staging
    app.config.from_object(config_class)
 
    init_extensions(app)                     # core/extensions.py
    init_db(app)                             # core/database.py
 
    from celery_app import init_celery       # deferred: avoid circular import
    init_celery(app)
 
    register_blueprints(app)                 # ~30 modules
    register_tenant_middleware(app)          # before_request tenant resolution
    register_error_handlers(app)
    register_health_check(app)
 
    if issubclass(config_class, ProductionConfig):
        config_class.init_app(app)           # prod/staging shared validation
 
    # start_memory_monitor(app) — RSS/VMS/CPU every 30s
    return app

Blueprint registration

Every module exposes a blueprint that is registered in register_blueprints() with an /api/<resource> prefix:

from modules.students.routes import students_bp
app.register_blueprint(students_bp, url_prefix='/api/students')

Representative prefixes (full list lives in register_blueprints()):

Others follow the same shape: academics, academic-terms, holidays, devices, school-units, programmes, grades, class-subjects, subject-contexts, mediums, religions, school-setup, sub-admins.

Module anatomy

Every domain module under server/modules/<name>/ follows the same layout:

server/modules/<name>/
├── __init__.py
├── models.py       ORM models (extend TenantBaseModel for tenant data)
├── routes.py       Flask blueprint + route handlers, decorators
├── services.py     Business logic (no raw DB access from routes)
└── schemas.py      (optional) request validation

The route handler stays thin: apply decorators, parse/validate input, delegate to a service, wrap the result in the response envelope. Business logic and multi-write transactions live in services.py.

Core infrastructure

server/core/ holds cross-cutting machinery shared by every module: database.py, models.py, tenant.py, branch_scope.py, extensions.py, feature_flags.py, plan_features.py, cache.py, validation.py, and the decorators/ package.

TenantBaseModel & automatic query scoping

All tenant-scoped models inherit TenantBaseModel:

class TenantBaseModel(db.Model):
    __abstract__ = True
    id = Column(Integer, primary_key=True)
    tenant_id = Column(Integer, ForeignKey('tenants.id'), nullable=False, index=True)
    created_at = Column(DateTime(timezone=True), default=func.now(), nullable=False)
    updated_at = Column(DateTime(timezone=True), default=func.now(), onupdate=func.now())

core/database.py registers a SQLAlchemy do_orm_execute listener that injects WHERE tenant_id = g.tenant_id into every ORM SELECT touching a TenantBaseModel subclass, via with_loader_criteria. It is a no-op outside a request context or when g.tenant_id is unset (platform / pre-auth paths).

This replaced an earlier before_compile hook that silently dropped its WHERE clause under LIMIT/OFFSET — so paginated lists leaked other tenants’ rows while count() scoped correctly. with_loader_criteria is applied at execution time and is safe under LIMIT/OFFSET, JOINs, and subqueries. It scopes SELECTs only — bulk UPDATE/DELETE must still filter tenant_id explicitly. Regression test: tests/test_tenant_scope_pagination_isolation.py.

Tenant resolution (core/tenant.py)

Registered as a before_request hook for all /api/* routes except health, OPTIONS, and platform routes. Resolution priority:

1. Request body field tenant_id or tenantId
2. Request body field subdomain
3. X-Tenant-ID header
4. Subdomain from the Host header (e.g. school.nexschool.com → school)
5. DEFAULT_TENANT_SUBDOMAIN env var (development fallback)

A suspended tenant returns 403; an unknown tenant returns 404. On success it sets g.tenant_id for route handlers and the query-scoping listener.

Branch scope (core/branch_scope.py)

Per-sub-admin branch (school-unit) scoping. Rule: no UserSchoolUnit rows for a user = unrestricted (all branches); rows present = restricted to that set; platform admins are always unrestricted.

• get_allowed_unit_ids() -> set|None / get_allowed_class_ids() -> set|None — resolvers cached on g (sentinel so None/unrestricted isn’t recomputed). None means unrestricted.
• Asserts (raise BranchForbidden → 403, no-op when unrestricted): assert_unit_allowed, assert_class_allowed, assert_student_allowed. A classless student fails closed for restricted users; a missing class/student is deferred to the caller’s 404 handling.
• List filters (subquery-based, no-op when unrestricted): filter_classes_by_branch, filter_by_class_ids(query, class_fk_column), filter_students_by_branch, filter_fees_by_branch(query, student_fk_column).

BranchForbidden is registered in register_error_handlers (app.py) via register_branch_scope_error_handler.

Decorators (core/decorators/)

@auth_required
# Validates JWT, loads g.current_user. 401 if no token or expired.
 
@tenant_required
# Ensures g.tenant_id is set (tenant resolution succeeded). 403 if missing.
 
@require_permission('student.read')
# Checks g.current_user has the permission via their roles. 403 if missing.
 
@require_any_permission('student.read', 'student.read.class')
# User must hold at least one of the listed permissions.
 
@require_plan_feature('timetable')
# Checks the tenant's plan enables this feature. 403 if not in plan.
 
@require_setup_complete
# Checks tenant.is_setup_complete. 400 if setup not done.
 
@require_active_subscription
# Checks the tenant subscription is active. 402/403 if suspended/cancelled.

Plan feature keys are defined in core/plan_features.py (PLAN_FEATURE_KEYS); RBAC helpers live in core/decorators/rbac.py; platform-admin gating in core/decorators/platform.py (@platform_admin_required). A Redis-backed cache layer (core/cache.py) fronts hot lookups such as user permissions.

Configuration & connection-pool hardening (config/settings.py)

class BaseConfig:
    SQLALCHEMY_TRACK_MODIFICATIONS = False
    JWT_EXPIRY_MINUTES = 15
    REFRESH_TOKEN_EXPIRY_DAYS = 30
    # Per-request body cap; keep in sync with nginx client_max_body_size.
    MAX_CONTENT_LENGTH = int(os.getenv("MAX_CONTENT_LENGTH", str(64 * 1024 * 1024)))
    SQLALCHEMY_ENGINE_OPTIONS = {
        "pool_pre_ping": True,   # replace connections dropped while idle
        "pool_recycle": 280,     # env: DB_POOL_RECYCLE
        "pool_size": 5,          # env: DB_POOL_SIZE
        "max_overflow": 10,      # env: DB_MAX_OVERFLOW
        # statement_timeout (ms): kill a runaway/locked query to free its conn+worker.
        "connect_args": {"options": "-c statement_timeout=120000"},  # env: DB_STATEMENT_TIMEOUT_MS
    }

DevelopmentConfig sets DEBUG=True; ProductionConfig sets DEBUG=False; StagingConfig subclasses ProductionConfig. get_config() (in config/__init__.py) selects the class from config_name or FLASK_ENV.

Async (Celery)

Background work runs on Celery with a Redis broker. Tasks are wrapped in a ContextTask so they execute inside a Flask app context (giving them DB access):

# celery_worker.py
def make_celery(app):
    celery = Celery(app.import_name, broker=app.config['CELERY_BROKER_URL'])
    celery.conf.update(app.config)
    TaskBase = celery.Task
 
    class ContextTask(TaskBase):
        def __call__(self, *args, **kwargs):
            with app.app_context():
                return TaskBase.__call__(self, *args, **kwargs)
 
    celery.Task = ContextTask
    return celery

create_app calls init_celery(app) (deferred import from celery_app to avoid a circular import). Task modules live in server/tasks/ (finance.py, notifications.py, notification_dispatch.py, push_notifications.py). Periodic jobs use a Beat schedule:

CELERYBEAT_SCHEDULE = {
    'process-overdue-fees-daily': {
        'task': 'tasks.finance.process_overdue_fees_task',
        'schedule': 86400.0,  # every 24 hours
    }
}

Response envelope & errors

Every route returns a consistent { success, ... } envelope.

// Success
{ "success": true, "data": { } }

// List with pagination
{
  "success": true,
  "data": [ ],
  "pagination": { "page": 1, "pageSize": 20, "totalItems": 142, "totalPages": 8 }
}

// Error
{ "success": false, "error": "ValidationError", "message": "Human-readable description" }

Global error handlers

Registered in register_error_handlers() (app.py). Client errors (4xx) log at WARN; unhandled server errors log at ERROR with a stack trace.

Never leak raw exception text (shared/safe_error.py)

Route and service handlers must never return raw exception text (str(e), str(e.orig)) to the client — psycopg2 output leaks SQL, table/column names, and tenant_id values. For the broad/DB catch path use safe_error:

from shared.safe_error import safe_error
 
except Exception as e:
    return {"success": False, "error": safe_error(e, "Failed to create student")}

safe_error(exc, message=DEFAULT_MESSAGE) logs the real exception with traceback to the nexchool.service_errors logger and returns only the generic message. Keep raw str(e) only for except ValueError/RuntimeError validation messages and custom AppError subclasses, which carry deliberate user-facing text. For IntegrityError, inspect str(getattr(e, "orig", None) or e) for the constraint name, then fall back to safe_error.

Migrations (Flask-Migrate)

Schema changes use Flask-Migrate (Alembic). Migration files live in server/migrations/versions/; server/run_migrate.py is the migration runner.

flask db migrate -m "add <name> tables"   # autogenerate a revision
flask db upgrade                           # apply
flask db downgrade                         # revert one revision

Adding a new module end to end:

1. Create server/modules/<name>/.
2. Add models.py (extend TenantBaseModel).
3. Add routes.py (create the blueprint, add routes with decorators).
4. Add services.py (business logic).
5. Register the blueprint in register_blueprints() in app.py.
6. Create a migration: flask db migrate -m "add <name> tables".
7. Add RBAC permissions in scripts/seed_rbac.py.

Related

• /architecture/overview
• /architecture/services
• /engineering/database-schema
• /engineering/adding-a-feature