CourseSpaceBackend — internal developer documentation

This README is intended for the developers working on the CourseSpace backend. It contains a concise overview of the project, repo layout, local development steps, testing guidance, and operational notes that help contributors get productive quickly.

• CourseSpaceBackend — internal developer documentation
  • Project overview
  • Repo layout (important directories)
  • Quick start (developer)
  • Local development (detailed)
  • Running and debugging
  • Database & migrations
  • Tests and fixtures
  • Authentication & authorization notes
  • Policy / permission system
  • Document storage (S3) and testing considerations
  • CI / release notes
  • Troubleshooting & tips
  • Adding a new feature (checklist)
  • Key architectural patterns

---

Project overview

• Web framework: FastAPI
• ORM: SQLAlchemy (Declarative models in src/models.py and subpackages)
• Database: PostgreSQL (migrations via Alembic)
• Background tasks: APScheduler (via SchedulerManager singleton)
• Storage: S3-compatible object storage for documents (via S3Manager singleton)
• Auth: Firebase Authentication with Bearer tokens (OAuth2PasswordBearer)
• Real-time: WebSocket support for poll updates (via PollWSManager singleton)
• AI: RAG pipeline with document chunking/embedding and LLM integration (via AIModelManager singleton)

The API serves course, thread, comment, document, folder, announcement, calendar, assignment, poll, prompt, notification, report, award, point, and user functionality used by the frontend apps.

Repo layout (important directories)

• src/ – application package
  • main.py – FastAPI app creation, router registration, lifespan (migrations, scheduler)
  • models.py, schemas.py, services.py (and subpackages) – domain code
  • auth/, course/, thread/, comment/, document/, folder/, announcement/, assignment/, calendar/, prompt/, user/, policy/, poll/, notification/, report/, award/, point/, scraper/, scheduler/, admin/, dev/ – feature modules
  • dependencies.py – unified Context dependency (combines db + user)
  • configs.py – environment-driven configuration
• tests/ – unit/integration tests, fixtures and builders
• scripts/ – helper scripts (start, setup)
• alembic/ – database migration configs and versions
• Dockerfile, compose.yaml – container and local compose definitions

Quick start (developer)

Prereqs: Docker + Docker Compose, Python 3.11+, git, VS Code

1. Copy environment variables:

   cp .env.example .env
   # edit .env to set secrets, DB urls, S3 config, etc.

2. Reopen this repository in the VS Code devcontainer when prompted.

3. Run the application using Ctrl + Shift + B or Cmd + Shift + B. Alternatively, run:

   ./scripts/start.sh

4. Open the API docs locally:

   http://localhost:8000/docs
   

5. Adminer / DB UI:

   http://localhost:8080
   

Local development (detailed)

• Use the devcontainer in VS Code for consistent development environment. Reopen repository in container when prompted.
• Dependencies are installed automatically in the devcontainer with scripts/setup.sh.
• Run the application with Ctrl + Shift + B or Cmd + Shift + B.
• The webserver runs on port 8000 by default. Modify compose.yaml or .env to change ports.
• When making schema or model changes remember to:
  • Create an Alembic revision: alembic revision --autogenerate -m "describe change"
  • Apply the migration in dev manually: alembic upgrade head
  • Or run the application which automatically applies migrations on startup.

Running and debugging

• Use the FastAPI interactive docs at /docs (Swagger) for quick API exploration.
• Use breakpoints and the Python debugger in VS Code. The devcontainer provides typical Python tooling.
• For local logs and troubleshooting, logs are written based on logging.conf; run with -s in tests to see output.

Database & migrations

• Migrations are in src/alembic/versions/ and follow the naming scheme YYYY_MM_DD_HHMM-<hash>_description.py.

• Migrations auto-apply on app startup via initialise_database() in the lifespan context.

• To generate a migration:

  alembic revision --autogenerate -m "describe change"

• To apply migrations manually (optional, as they auto-apply):

  alembic upgrade head

CI/CD applies migrations during deployment — make sure migrations are atomic and tested.

Tests and fixtures

• Tests live under tests/ and are organized by feature (auth, thread, comment, course, etc.).

• Builders (tests/**/builders.py) create test data objects WITHOUT DB persistence:

  thread_data = ThreadBuilder(status=ThreadStatus.ACTIVE).build()  # Returns ThreadData

  • Builders use method chaining for optional fields (e.g., .as_public(), .disallow_comments())

• Factory fixtures (tests/**/fixtures.py) create actual DB records:

  thread = create_thread_factory(user, course)  # Committed to test DB

• All tests/**/fixtures.py are auto-loaded as pytest plugins (see tests/conftest.py).

• Tests run in rolled-back transactions for isolation.

• To run tests:

  pytest -q           # All tests, quiet output
  pytest -k test_name # Specific test
  pytest -s           # Show print output

• Test assertions (from tests/assertions.py):

  • assert_response(response, 200) - check status, return JSON
  • assert_error_response(response, 404, "THR", "does not exist") - validate error format
  • assert_response_data(response_data, expected_data) - compare response fields (use "<EXISTS>" for fields that just need to exist)

• Auth mocking: Use mock_firebase_auth(client, user) fixture to mock Firebase token verification. This patches verify_id_token and sets Authorization header.

Authentication & authorization notes

• Authentication: Firebase Authentication with Bearer tokens (OAuth2PasswordBearer scheme).

• Routes use unified Context dependency combining DB session and authenticated user:

  from src.dependencies import Context, get_context
  
  @router.get("/{id}")
  @require_permission("resource", "read", "id")
  def get_resource(id: UUID, context: Context = Depends(get_context)):
      # Access: context.db (Session), context.user (User)

• WebSocket routes: Use get_ws_context instead (reads auth from WebSocket headers):

  from src.dependencies import get_ws_context
  
  @router.websocket("/{id}/ws")
  @require_permission("poll", "read", "id")
  async def poll_websocket(websocket: WebSocket, id: UUID, context: Context = Depends(get_ws_context)):
      await wsManager.connect(str(id), websocket)

• Never use separate get_db and get_current_user dependencies — always use Context.

• Key files:

  • src/auth/dependencies.py — Firebase token validation, get_current_user, get_current_ws_user
  • src/dependencies.py — Context, get_context, and get_ws_context (WebSocket)

• User records auto-created via get_or_create_user on first authentication.

• Token verification checks: revocation status, user disabled status, token validity.

Policy / permission system

• All protected routes use @require_permission(resource_type, action, resource_id_param) from src/policy/decorators.py.

• Authorization lives in policies—never in routes or services.

• Decorator extracts context from route kwargs and validates before execution.

• Policies receive PolicyContext(db, **route_kwargs) which provides access to DB session and all route parameters via .get(key) method.

• Policies can delegate (e.g., ThreadPolicy checks CoursePolicy for course access).

• To add a new resource policy:

  1. Create src/<feature>/policies.py implementing BasePolicy.can_perform_action(user, action, resource, context) where context is a PolicyContext instance
  2. Register at end of src/policy/managers.py:

     PolicyManager.register("resource", ResourcePolicy)

  3. Decorate route:

     @require_permission("thread", "update", "id")
     def update_thread(id: UUID, context: Context = Depends(get_context)): ...

Document storage (S3) and testing considerations

• Production uses S3-compatible storage. Tests should not upload to production S3.
• For tests prefer mocking S3 calls (see tests/document/fixtures.py) or use moto / a local S3-compatible server.
• The document fixtures upload and then clean up S3 objects — ensure tests track created object names for cleanup.

CI / release notes

• CI runs unit tests, linting and static analysis. Keep changes that affect public APIs backward compatible or version them.
• When bumping dependencies, update requirements/ and run full test suite locally to ensure compatibility.

Troubleshooting & tips

• DB connection errors: ensure Docker compose is up and that .env has correct DB URL.
• Alembic errors: check env.py and alembic.ini for correct SQLALCHEMY URL mapping.
• S3 issues: verify credentials and endpoint; run S3-integrated tests with a mock.
• HTTP errors: check error code and message for clues, refer to output.log for server-side details.

Adding a new feature (checklist)

1. Create src/<feature>/models.py - SQLAlchemy models (inherit from Base in src/models.py)
2. Create src/<feature>/schemas.py - Pydantic request/response models
3. Create src/<feature>/services.py - business logic with db as first param
4. Create src/<feature>/policies.py - implement BasePolicy.can_perform_action
5. Create src/<feature>/routers.py - FastAPI routes with @require_permission
6. Register policy: PolicyManager.register("feature", FeaturePolicy) at end of src/policy/managers.py
7. Register router: app.include_router(feature_router) in src/main.py
8. Create migration: alembic revision --autogenerate -m "add feature"
9. Create tests/<feature>/builders.py, fixtures.py, test_<feature>.py

Key architectural patterns

• Exception handling: Use typed exceptions with 3-letter module codes:

  raise NotFoundException("THR", f"Thread [{id}] does not exist.")
  raise BadRequestException("CMT", f"Comment [{id}] is deleted.")

  Format: <CODE>::<Message>. Codes: THR, CMT, AWD, COR, DOC, FLD, ANN, PNT, ASM, CLD, PMT, PLC, AUT, NTF, POL, RPT, USR, SCR.

• Soft deletes: Use is_deleted flag or status enum (active/archived/deleted):

  thread = get_existing_thread(db, id)  # Check existence only
  thread = get_existing_active_thread(db, id)  # Check existence AND active status

• Singleton managers: S3Manager, SchedulerManager, PollWSManager, AIModelManager use metaclass=Singleton from src/metaclasses.py

  • Example: SchedulerManager() always returns same instance
  • Initialized in src/main.py lifespan
  • SchedulerManager runs scheduled jobs:
    • broadcast_poll_updates - broadcasts poll updates via WebSocket
    • check_active_polls_expiration - marks expired polls as closed
    • process_document_chunks - processes documents for RAG (chunking, embedding)
    • generate_prompts_for_threads - generates AI assistant prompts for threads

• AI/RAG Pipeline: Retrieval-Augmented Generation for AI assistant features (requires course setting enable_ai_assistant=True):

  • Document Processing (via DocumentProcesser singleton in src/prompt/managers.py):
    • Loads documents from S3 storage (supports: txt, pdf, docx, pptx)
    • Splits into chunks using RecursiveCharacterTextSplitter (default: 1000 chars, 200 overlap)
    • Creates embeddings via TextEmbedder (Google Generative AI Embeddings)
    • Stores chunks with embeddings in document_chunk table
  • Thread Prompt Generation (via AIModelManager singleton):
    • Finds threads eligible for AI assistance (based on course settings and thread age)
    • Embeds thread title + content as query vector
    • Retrieves similar document chunks from same course (vector similarity search)
    • Generates contextual prompt using LLM with retrieved chunks as context
    • Stores prompt in prompt table linked to thread
  • Scheduled Processing:
    • process_document_chunks - runs every N minutes, processes documents with chunk_processing_status=PENDING
    • generate_prompts_for_threads - runs every N minutes, generates prompts for threads meeting age threshold
  • Key files:
    • src/prompt/managers.py - DocumentProcesser, TextEmbedder, AIModelManager classes
    • src/prompt/configs.py - RAG configuration (chunk size, embedding model, prompt model)
    • src/document/models.py - DocumentChunk model with embedding column
    • src/scheduler/managers.py - scheduled job definitions

• Base model utilities: All SQLAlchemy models inherit from Base in src/models.py:

  • .dict() - convert model to dictionary of column values
  • __repr__() - detailed string representation with all attributes
  • __str__() - string representation via .dict()

  thread = Thread(title="Test", content="Content")
  thread_dict = thread.dict()  # Returns {'id': ..., 'title': 'Test', ...}

Last updated: 28-Feb-2026