-
Notifications
You must be signed in to change notification settings - Fork 0
backend
Mile Shi edited this page May 26, 2025
·
1 revision
This guide covers backend development for the Intelligent IDE project, built with FastAPI and modern Python practices.
The backend is a FastAPI-based REST API server that provides:
- User authentication and management
- Course creation and management
- File storage and organization
- Real-time collaboration features
- Assignment and homework systems
- Chat functionality
- Framework: FastAPI 0.104+
- Language: Python 3.11+
- Database: PostgreSQL 13+
- Cache: Redis 6+
- ORM: SQLAlchemy 2.0+
- Validation: Pydantic 2.0+
- Authentication: JWT tokens
- Containerization: Docker
Based on the actual backend structure:
backend/intellide/
├── main.py # FastAPI application entry point
├── config.py # Configuration management
├── cache/ # Redis cache implementation
│ ├── __init__.py
│ ├── cache.py # Cache operations
│ └── startup.py # Cache initialization
├── database/ # Database layer
│ ├── __init__.py
│ ├── database.py # Database connection
│ ├── model.py # SQLAlchemy models
│ └── startup.py # Database initialization
├── docker/ # Docker management
│ ├── __init__.py
│ └── startup.py # Docker service startup
├── routers/ # API route handlers
│ ├── __init__.py
│ ├── course.py # Course management
│ ├── course_chat.py # Chat functionality
│ ├── course_directory.py # File management
│ ├── course_directory_entry.py # File entries
│ ├── course_student.py # Student enrollment
│ ├── surprise.py # Special features
│ └── user.py # User management
├── storage/ # File storage
│ ├── __init__.py
│ ├── startup.py # Storage initialization
│ └── storage.py # Storage operations
├── tests/ # Test suite
│ ├── __init__.py
│ ├── conftest.py # Test configuration
│ ├── pytest.ini # Pytest settings
│ ├── test_course.py # Course tests
│ ├── test_user.py # User tests
│ └── utils.py # Test utilities
└── utils/ # Utility functions
├── __init__.py
├── auth.py # Authentication helpers
├── email.py # Email services
├── path.py # Path utilities
├── response.py # Response formatting
└── websocket.py # WebSocket utilities
- Python 3.11+: Ensure Python is installed
- Docker: Required for PostgreSQL and Redis
- pip: For package management
-
Navigate to Backend Directory
cd backend/deploy -
Install Dependencies
pip install -r requirements.txt
-
Configure Docker
- Enable Docker daemon exposure on localhost:2375
- Start Docker Desktop
-
Start Services
# macOS/Linux sh backend.sh # Windows backend.bat
-
Verify Installation
- Open http://localhost:8080/docs for Swagger UI
- Check http://localhost:8080/redoc for ReDoc
Create Pydantic models for request/response validation:
# In routers/your_feature.py
from pydantic import BaseModel
from typing import Optional
class FeatureCreate(BaseModel):
name: str
description: Optional[str] = None
class FeatureResponse(BaseModel):
id: int
name: str
description: Optional[str]
created_at: datetime
class Config:
from_attributes = TrueDefine SQLAlchemy models in database/model.py:
# Add to database/model.py
class Feature(Base):
__tablename__ = "features"
id = Column(Integer, primary_key=True, index=True)
name = Column(String, index=True)
description = Column(String, nullable=True)
created_at = Column(DateTime, default=datetime.utcnow)
# Add relationships if needed
user_id = Column(Integer, ForeignKey("users.id"))
user = relationship("User", back_populates="features")Create API endpoints in routers/:
# routers/your_feature.py
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from database.database import get_db
from utils.auth import get_current_user
router = APIRouter(
prefix="/api/feature",
tags=["features"]
)
@router.post("/", response_model=FeatureResponse)
async def create_feature(
feature: FeatureCreate,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user)
):
# Implementation logic
db_feature = Feature(**feature.dict(), user_id=current_user.id)
db.add(db_feature)
db.commit()
db.refresh(db_feature)
return db_feature
@router.get("/{feature_id}", response_model=FeatureResponse)
async def get_feature(
feature_id: int,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user)
):
feature = db.query(Feature).filter(Feature.id == feature_id).first()
if not feature:
raise HTTPException(status_code=404, detail="Feature not found")
return featureAdd the router to main.py:
# In main.py
from routers import your_feature
app.include_router(your_feature.router)The system uses JWT tokens for authentication:
# utils/auth.py implementation
from jose import JWTError, jwt
from datetime import datetime, timedelta
def create_access_token(data: dict):
to_encode = data.copy()
expire = datetime.utcnow() + timedelta(hours=24)
to_encode.update({"exp": expire})
encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
return encoded_jwt
def get_current_user(
token: str = Depends(oauth2_scheme),
db: Session = Depends(get_db)
):
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
username: str = payload.get("sub")
if username is None:
raise HTTPException(status_code=401, detail="Invalid token")
except JWTError:
raise HTTPException(status_code=401, detail="Invalid token")
user = db.query(User).filter(User.username == username).first()
if user is None:
raise HTTPException(status_code=401, detail="User not found")
return user@router.get("/protected")
async def protected_route(
current_user: User = Depends(get_current_user)
):
return {"message": f"Hello {current_user.username}"}# database/database.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
SQLALCHEMY_DATABASE_URL = "postgresql://user:password@localhost/dbname"
engine = create_engine(SQLALCHEMY_DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()# Example CRUD operations
def create_course(db: Session, course: CourseCreate, user_id: int):
db_course = Course(**course.dict(), creator_id=user_id)
db.add(db_course)
db.commit()
db.refresh(db_course)
return db_course
def get_courses(db: Session, skip: int = 0, limit: int = 100):
return db.query(Course).offset(skip).limit(limit).all()
def get_course(db: Session, course_id: int):
return db.query(Course).filter(Course.id == course_id).first()
def update_course(db: Session, course_id: int, course: CourseUpdate):
db.query(Course).filter(Course.id == course_id).update(course.dict())
db.commit()
return db.query(Course).filter(Course.id == course_id).first()# cache/cache.py
import redis
import json
from typing import Any, Optional
class CacheService:
def __init__(self):
self.redis_client = redis.Redis(host='localhost', port=6379, db=0)
def set(self, key: str, value: Any, expire: int = 3600):
"""Set a value in cache with expiration."""
serialized = json.dumps(value)
self.redis_client.setex(key, expire, serialized)
def get(self, key: str) -> Optional[Any]:
"""Get a value from cache."""
value = self.redis_client.get(key)
if value:
return json.loads(value)
return None
def delete(self, key: str):
"""Delete a key from cache."""
self.redis_client.delete(key)from cache.cache import CacheService
cache = CacheService()
@router.get("/courses/{course_id}")
async def get_course_cached(course_id: int, db: Session = Depends(get_db)):
# Check cache first
cached_course = cache.get(f"course:{course_id}")
if cached_course:
return cached_course
# If not in cache, query database
course = db.query(Course).filter(Course.id == course_id).first()
if course:
# Cache the result
cache.set(f"course:{course_id}", course.dict(), expire=1800)
return course# storage/storage.py
import os
import uuid
from fastapi import UploadFile
class StorageService:
def __init__(self, base_path: str = "./storage"):
self.base_path = base_path
os.makedirs(base_path, exist_ok=True)
async def save_file(self, file: UploadFile, subdir: str = "") -> str:
"""Save uploaded file and return file path."""
file_id = str(uuid.uuid4())
dir_path = os.path.join(self.base_path, subdir)
os.makedirs(dir_path, exist_ok=True)
file_path = os.path.join(dir_path, file_id)
with open(file_path, "wb") as f:
content = await file.read()
f.write(content)
return file_id
def get_file_path(self, file_id: str, subdir: str = "") -> str:
"""Get full path for a file."""
return os.path.join(self.base_path, subdir, file_id)@router.post("/upload")
async def upload_file(
file: UploadFile = File(...),
course_id: int = Form(...),
current_user: User = Depends(get_current_user),
db: Session = Depends(get_db)
):
storage = StorageService()
file_id = await storage.save_file(file, f"course_{course_id}")
# Save file metadata to database
db_file = CourseDirectoryEntry(
course_id=course_id,
name=file.filename,
file_id=file_id,
uploaded_by=current_user.id
)
db.add(db_file)
db.commit()
return {"file_id": file_id, "filename": file.filename}# utils/websocket.py
from typing import List
from fastapi import WebSocket
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 send_personal_message(self, message: str, websocket: WebSocket):
await websocket.send_text(message)
async def broadcast(self, message: str):
for connection in self.active_connections:
await connection.send_text(message)
manager = ConnectionManager()@router.websocket("/ws/course/{course_id}")
async def websocket_endpoint(
websocket: WebSocket,
course_id: int
):
await manager.connect(websocket)
try:
while True:
data = await websocket.receive_text()
# Process message and broadcast to course participants
await manager.broadcast(f"Course {course_id}: {data}")
except WebSocketDisconnect:
manager.disconnect(websocket)# tests/conftest.py
import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from fastapi.testclient import TestClient
from main import app
from database.database import get_db, Base
SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False})
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
def override_get_db():
try:
db = TestingSessionLocal()
yield db
finally:
db.close()
app.dependency_overrides[get_db] = override_get_db
@pytest.fixture
def client():
Base.metadata.create_all(bind=engine)
with TestClient(app) as c:
yield c
Base.metadata.drop_all(bind=engine)# tests/test_user.py
def test_create_user(client):
response = client.post(
"/api/user/register",
json={
"username": "testuser",
"email": "testuser@testdomain.local",
"password": "testpassword",
"role": "student"
}
)
assert response.status_code == 200
data = response.json()
assert data["username"] == "testuser"
assert "access_token" in data
def test_login_user(client):
# First create a user
client.post("/api/user/register", json={...})
# Then test login
response = client.post(
"/api/user/login",
json={
"username": "testuser",
"password": "testpassword"
}
)
assert response.status_code == 200
data = response.json()
assert "access_token" in data# Run all tests
python -m pytest
# Run specific test file
python -m pytest tests/test_user.py
# Run with coverage
python -m pytest --cov=intellide tests/# config.py
import os
from pydantic import BaseSettings
class Settings(BaseSettings):
database_url: str = "postgresql://user:password@localhost/intellide"
redis_url: str = "redis://localhost:6379"
secret_key: str = "your-secret-key"
email_host: str = "smtp.gmail.com"
email_port: int = 587
email_username: str = ""
email_password: str = ""
class Config:
env_file = ".env"
settings = Settings()from fastapi import HTTPException
from fastapi.responses import JSONResponse
@app.exception_handler(HTTPException)
async def http_exception_handler(request, exc):
return JSONResponse(
status_code=exc.status_code,
content={"message": exc.detail, "status": "error"}
)# Use select_related for eager loading
def get_course_with_students(db: Session, course_id: int):
return db.query(Course)\
.options(joinedload(Course.students))\
.filter(Course.id == course_id)\
.first()
# Use pagination for large datasets
def get_courses_paginated(db: Session, skip: int = 0, limit: int = 20):
return db.query(Course)\
.offset(skip)\
.limit(limit)\
.all()from functools import lru_cache
@lru_cache(maxsize=128)
def get_course_metadata(course_id: int):
# Expensive computation
return compute_metadata(course_id)This comprehensive backend development guide provides the foundation for contributing to and extending the Intelligent IDE backend system.
🏠 Home
- Getting Started
- Installation Guide
- Authentication
- Course Management
- Collaborative Editing
- Assignments
- Notebook Features
- File Management
- Troubleshooting
- Setup & Development
- Architecture Overview
- Backend Development
- Frontend Development
- API Reference
- Contributing
- Deployment