Implement comprehensive school portal API with FastAPI

- Complete authentication system with JWT and role-based access control
- User management for Admin, Teacher, Student, and Parent roles
- Student management with CRUD operations
- Class management and assignment system
- Subject and grade tracking functionality
- Daily attendance marking and viewing
- Notification system for announcements
- SQLite database with Alembic migrations
- Comprehensive API documentation with Swagger/ReDoc
- Proper project structure with services, models, and schemas
- Environment variable configuration
- CORS support and security features

🤖 Generated with BackendIM

Co-Authored-By: BackendIM <noreply@anthropic.com>
This commit is contained in:
Automated Action 2025-06-25 13:31:56 +00:00
parent ab56cbf293
commit 5a02fb8b1f
49 changed files with 1780 additions and 2 deletions

153
README.md
View File

@ -1,3 +1,152 @@
# FastAPI Application # School Portal API
This is a FastAPI application bootstrapped by BackendIM, the AI-powered backend generation platform. A comprehensive school management system built with Python FastAPI, supporting multiple user roles and complete student lifecycle management.
## Features
### User Roles
- **Admin**: Full system access, user management, class management
- **Teacher**: Student management, grade tracking, attendance marking, notifications
- **Student**: View grades, attendance, and notifications
- **Parent**: View child's grades, attendance, and notifications
### Core Functionality
- **Authentication**: JWT-based login and registration for all user roles
- **Student Management**: Create, update, and delete student profiles
- **Class Management**: Assign students to classes and teachers to classes
- **Subject & Grade Tracking**: Teachers record grades per subject, students/parents view them
- **Attendance**: Daily attendance marking and viewing
- **Notifications**: Announcements and messages to students and parents
## Project Structure
```
schoolportalapi/
├── app/
│ ├── api/
│ │ └── v1/
│ │ ├── endpoints/
│ │ │ ├── auth.py # Authentication endpoints
│ │ │ ├── users.py # User management
│ │ │ ├── classes.py # Class management
│ │ │ ├── subjects.py # Subject management
│ │ │ ├── grades.py # Grade tracking
│ │ │ ├── attendance.py # Attendance management
│ │ │ └── notifications.py # Notification system
│ │ └── api.py # API router
│ ├── core/
│ │ ├── config.py # Configuration settings
│ │ └── security.py # JWT and password utilities
│ ├── db/
│ │ ├── base.py # SQLAlchemy base
│ │ └── session.py # Database session
│ ├── models/ # SQLAlchemy models
│ │ ├── user.py
│ │ ├── class_model.py
│ │ ├── subject.py
│ │ ├── grade.py
│ │ ├── attendance.py
│ │ └── notification.py
│ ├── schemas/ # Pydantic schemas
│ │ ├── user.py
│ │ ├── class_schema.py
│ │ ├── subject.py
│ │ ├── grade.py
│ │ ├── attendance.py
│ │ ├── notification.py
│ │ └── token.py
│ └── services/ # Business logic layer
│ ├── base.py
│ ├── user.py
│ ├── class_service.py
│ ├── subject.py
│ ├── grade.py
│ ├── attendance.py
│ └── notification.py
├── alembic/ # Database migrations
│ ├── versions/
│ └── env.py
├── main.py # FastAPI application entry point
├── requirements.txt # Python dependencies
└── alembic.ini # Alembic configuration
```
## Setup Instructions
### Prerequisites
- Python 3.8+
- pip
### Installation
1. **Clone the repository**
```bash
git clone <repository-url>
cd schoolportalapi
```
2. **Create virtual environment**
```bash
python -m venv venv
source venv/bin/activate # On Windows: venv\Scripts\activate
```
3. **Install dependencies**
```bash
pip install -r requirements.txt
```
4. **Set environment variables**
Create a `.env` file or set the following environment variables:
```bash
# Required Environment Variables
SECRET_KEY=your-secret-key-change-in-production
DATABASE_URL=sqlite:////app/storage/db/db.sqlite
FIRST_SUPERUSER_EMAIL=admin@schoolportal.com
FIRST_SUPERUSER_PASSWORD=admin123
```
5. **Create storage directory**
```bash
mkdir -p /app/storage/db
```
6. **Run database migrations**
```bash
alembic upgrade head
```
7. **Start the application**
```bash
uvicorn main:app --reload --host 0.0.0.0 --port 8000
```
## API Documentation
Once the application is running, access the interactive API documentation:
- **Swagger UI**: http://localhost:8000/docs
- **ReDoc**: http://localhost:8000/redoc
- **OpenAPI JSON**: http://localhost:8000/openapi.json
## Environment Variables
| Variable | Description | Default |
|----------|-------------|---------|
| `SECRET_KEY` | JWT secret key | `your-secret-key-change-in-production` |
| `DATABASE_URL` | Database connection URL | `sqlite:////app/storage/db/db.sqlite` |
| `FIRST_SUPERUSER_EMAIL` | Initial admin email | `admin@schoolportal.com` |
| `FIRST_SUPERUSER_PASSWORD` | Initial admin password | `admin123` |
## Security Features
- JWT-based authentication
- Password hashing using bcrypt
- Role-based access control
- CORS configuration for cross-origin requests
- Input validation using Pydantic schemas
## Health Check
The application provides a health check endpoint:
- `GET /health` - Returns application health status

41
alembic.ini Normal file
View File

@ -0,0 +1,41 @@
[alembic]
script_location = alembic
prepend_sys_path = .
version_path_separator = os
sqlalchemy.url = sqlite:////app/storage/db/db.sqlite
[post_write_hooks]
[loggers]
keys = root,sqlalchemy,alembic
[handlers]
keys = console
[formatters]
keys = generic
[logger_root]
level = WARN
handlers = console
qualname =
[logger_sqlalchemy]
level = WARN
handlers =
qualname = sqlalchemy.engine
[logger_alembic]
level = INFO
handlers =
qualname = alembic
[handler_console]
class = StreamHandler
args = (sys.stderr,)
level = NOTSET
formatter = generic
[formatter_generic]
format = %(levelname)-5.5s [%(name)s] %(message)s
datefmt = %H:%M:%S

49
alembic/env.py Normal file
View File

@ -0,0 +1,49 @@
from logging.config import fileConfig
from sqlalchemy import engine_from_config
from sqlalchemy import pool
from alembic import context
import sys
from pathlib import Path
sys.path.append(str(Path(__file__).resolve().parent.parent))
from app.db.base import Base
config = context.config
if config.config_file_name is not None:
fileConfig(config.config_file_name)
target_metadata = Base.metadata
def run_migrations_offline() -> None:
url = config.get_main_option("sqlalchemy.url")
context.configure(
url=url,
target_metadata=target_metadata,
literal_binds=True,
dialect_opts={"paramstyle": "named"},
)
with context.begin_transaction():
context.run_migrations()
def run_migrations_online() -> None:
connectable = engine_from_config(
config.get_section(config.config_ini_section),
prefix="sqlalchemy.",
poolclass=pool.NullPool,
)
with connectable.connect() as connection:
context.configure(
connection=connection, target_metadata=target_metadata
)
with context.begin_transaction():
context.run_migrations()
if context.is_offline_mode():
run_migrations_offline()
else:
run_migrations_online()

23
alembic/script.py.mako Normal file
View File

@ -0,0 +1,23 @@
"""${message}
Revision ID: ${up_revision}
Revises: ${down_revision | comma,n}
Create Date: ${create_date}
"""
from alembic import op
import sqlalchemy as sa
${imports if imports else ""}
revision = '${up_revision}'
down_revision = ${repr(down_revision)}
branch_labels = ${repr(branch_labels)}
depends_on = ${repr(depends_on)}
def upgrade() -> None:
${upgrades if upgrades else "pass"}
def downgrade() -> None:
${downgrades if downgrades else "pass"}

View File

@ -0,0 +1,147 @@
"""Initial migration
Revision ID: 001
Revises:
Create Date: 2024-01-01 00:00:00.000000
"""
from alembic import op
import sqlalchemy as sa
revision = '001'
down_revision = None
branch_labels = None
depends_on = None
def upgrade() -> None:
op.create_table('classes',
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('name', sa.String(), nullable=False),
sa.Column('grade_level', sa.String(), nullable=False),
sa.Column('academic_year', sa.String(), nullable=False),
sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('(CURRENT_TIMESTAMP)'), nullable=True),
sa.Column('updated_at', sa.DateTime(timezone=True), nullable=True),
sa.PrimaryKeyConstraint('id')
)
op.create_index(op.f('ix_classes_id'), 'classes', ['id'], unique=False)
op.create_table('users',
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('email', sa.String(), nullable=False),
sa.Column('hashed_password', sa.String(), nullable=False),
sa.Column('first_name', sa.String(), nullable=False),
sa.Column('last_name', sa.String(), nullable=False),
sa.Column('role', sa.Enum('admin', 'teacher', 'student', 'parent', name='userrole'), nullable=False),
sa.Column('is_active', sa.Boolean(), nullable=True),
sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('(CURRENT_TIMESTAMP)'), nullable=True),
sa.Column('updated_at', sa.DateTime(timezone=True), nullable=True),
sa.Column('parent_id', sa.Integer(), nullable=True),
sa.Column('class_id', sa.Integer(), nullable=True),
sa.ForeignKeyConstraint(['class_id'], ['classes.id'], ),
sa.ForeignKeyConstraint(['parent_id'], ['users.id'], ),
sa.PrimaryKeyConstraint('id')
)
op.create_index(op.f('ix_users_email'), 'users', ['email'], unique=True)
op.create_index(op.f('ix_users_id'), 'users', ['id'], unique=False)
op.create_table('subjects',
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('name', sa.String(), nullable=False),
sa.Column('code', sa.String(), nullable=False),
sa.Column('description', sa.String(), nullable=True),
sa.Column('class_id', sa.Integer(), nullable=False),
sa.Column('teacher_id', sa.Integer(), nullable=False),
sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('(CURRENT_TIMESTAMP)'), nullable=True),
sa.Column('updated_at', sa.DateTime(timezone=True), nullable=True),
sa.ForeignKeyConstraint(['class_id'], ['classes.id'], ),
sa.ForeignKeyConstraint(['teacher_id'], ['users.id'], ),
sa.PrimaryKeyConstraint('id'),
sa.UniqueConstraint('code')
)
op.create_index(op.f('ix_subjects_id'), 'subjects', ['id'], unique=False)
op.create_table('teacher_classes',
sa.Column('teacher_id', sa.Integer(), nullable=False),
sa.Column('class_id', sa.Integer(), nullable=False),
sa.ForeignKeyConstraint(['class_id'], ['classes.id'], ),
sa.ForeignKeyConstraint(['teacher_id'], ['users.id'], ),
sa.PrimaryKeyConstraint('teacher_id', 'class_id')
)
op.create_table('attendance',
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('student_id', sa.Integer(), nullable=False),
sa.Column('class_id', sa.Integer(), nullable=False),
sa.Column('date', sa.Date(), nullable=False),
sa.Column('is_present', sa.Boolean(), nullable=True),
sa.Column('remarks', sa.String(), nullable=True),
sa.Column('marked_by', sa.Integer(), nullable=False),
sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('(CURRENT_TIMESTAMP)'), nullable=True),
sa.Column('updated_at', sa.DateTime(timezone=True), nullable=True),
sa.ForeignKeyConstraint(['class_id'], ['classes.id'], ),
sa.ForeignKeyConstraint(['marked_by'], ['users.id'], ),
sa.ForeignKeyConstraint(['student_id'], ['users.id'], ),
sa.PrimaryKeyConstraint('id')
)
op.create_index(op.f('ix_attendance_id'), 'attendance', ['id'], unique=False)
op.create_table('grades',
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('student_id', sa.Integer(), nullable=False),
sa.Column('subject_id', sa.Integer(), nullable=False),
sa.Column('score', sa.Float(), nullable=False),
sa.Column('max_score', sa.Float(), nullable=False),
sa.Column('grade_type', sa.String(), nullable=False),
sa.Column('description', sa.String(), nullable=True),
sa.Column('graded_at', sa.DateTime(timezone=True), server_default=sa.text('(CURRENT_TIMESTAMP)'), nullable=True),
sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('(CURRENT_TIMESTAMP)'), nullable=True),
sa.Column('updated_at', sa.DateTime(timezone=True), nullable=True),
sa.ForeignKeyConstraint(['student_id'], ['users.id'], ),
sa.ForeignKeyConstraint(['subject_id'], ['subjects.id'], ),
sa.PrimaryKeyConstraint('id')
)
op.create_index(op.f('ix_grades_id'), 'grades', ['id'], unique=False)
op.create_table('notifications',
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('title', sa.String(), nullable=False),
sa.Column('message', sa.Text(), nullable=False),
sa.Column('sender_id', sa.Integer(), nullable=False),
sa.Column('notification_type', sa.String(), nullable=False),
sa.Column('priority', sa.String(), nullable=True),
sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('(CURRENT_TIMESTAMP)'), nullable=True),
sa.Column('updated_at', sa.DateTime(timezone=True), nullable=True),
sa.ForeignKeyConstraint(['sender_id'], ['users.id'], ),
sa.PrimaryKeyConstraint('id')
)
op.create_index(op.f('ix_notifications_id'), 'notifications', ['id'], unique=False)
op.create_table('notification_recipients',
sa.Column('notification_id', sa.Integer(), nullable=False),
sa.Column('user_id', sa.Integer(), nullable=False),
sa.Column('is_read', sa.Boolean(), nullable=True),
sa.Column('read_at', sa.DateTime(timezone=True), nullable=True),
sa.ForeignKeyConstraint(['notification_id'], ['notifications.id'], ),
sa.ForeignKeyConstraint(['user_id'], ['users.id'], ),
sa.PrimaryKeyConstraint('notification_id', 'user_id')
)
def downgrade() -> None:
op.drop_table('notification_recipients')
op.drop_index(op.f('ix_notifications_id'), table_name='notifications')
op.drop_table('notifications')
op.drop_index(op.f('ix_grades_id'), table_name='grades')
op.drop_table('grades')
op.drop_index(op.f('ix_attendance_id'), table_name='attendance')
op.drop_table('attendance')
op.drop_table('teacher_classes')
op.drop_index(op.f('ix_subjects_id'), table_name='subjects')
op.drop_table('subjects')
op.drop_index(op.f('ix_users_id'), table_name='users')
op.drop_index(op.f('ix_users_email'), table_name='users')
op.drop_table('users')
op.drop_index(op.f('ix_classes_id'), table_name='classes')
op.drop_table('classes')

0
app/__init__.py Normal file
View File

0
app/api/__init__.py Normal file
View File

62
app/api/deps.py Normal file
View File

@ -0,0 +1,62 @@
from typing import Generator
from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from sqlalchemy.orm import Session
from app.db.session import SessionLocal
from app.core import security
from app.models.user import User, UserRole
from app.services.user import user_service
security_scheme = HTTPBearer()
def get_db() -> Generator:
try:
db = SessionLocal()
yield db
finally:
db.close()
def get_current_user(
db: Session = Depends(get_db),
credentials: HTTPAuthorizationCredentials = Depends(security_scheme)
) -> User:
token = credentials.credentials
user_id = security.verify_token(token)
if not user_id:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials",
)
user = user_service.get(db, id=int(user_id))
if not user:
raise HTTPException(status_code=404, detail="User not found")
if not user.is_active:
raise HTTPException(status_code=400, detail="Inactive user")
return user
def get_current_active_user(
current_user: User = Depends(get_current_user),
) -> User:
if not current_user.is_active:
raise HTTPException(status_code=400, detail="Inactive user")
return current_user
def get_current_admin_user(
current_user: User = Depends(get_current_user),
) -> User:
if current_user.role != UserRole.ADMIN:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Not enough permissions"
)
return current_user
def get_current_teacher_or_admin(
current_user: User = Depends(get_current_user),
) -> User:
if current_user.role not in [UserRole.TEACHER, UserRole.ADMIN]:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Not enough permissions"
)
return current_user

0
app/api/v1/__init__.py Normal file
View File

11
app/api/v1/api.py Normal file
View File

@ -0,0 +1,11 @@
from fastapi import APIRouter
from app.api.v1.endpoints import auth, users, classes, subjects, grades, attendance, notifications
api_router = APIRouter()
api_router.include_router(auth.router, prefix="/auth", tags=["authentication"])
api_router.include_router(users.router, prefix="/users", tags=["users"])
api_router.include_router(classes.router, prefix="/classes", tags=["classes"])
api_router.include_router(subjects.router, prefix="/subjects", tags=["subjects"])
api_router.include_router(grades.router, prefix="/grades", tags=["grades"])
api_router.include_router(attendance.router, prefix="/attendance", tags=["attendance"])
api_router.include_router(notifications.router, prefix="/notifications", tags=["notifications"])

View File

View File

@ -0,0 +1,83 @@
from typing import Any, List
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from datetime import date
from app.api import deps
from app.models.user import User, UserRole
from app.schemas.attendance import Attendance, AttendanceCreate, AttendanceUpdate
from app.services.attendance import attendance_service
router = APIRouter()
@router.get("/", response_model=List[Attendance])
def read_attendance(
db: Session = Depends(deps.get_db),
skip: int = 0,
limit: int = 100,
student_id: int = None,
class_id: int = None,
date: date = None,
current_user: User = Depends(deps.get_current_active_user),
) -> Any:
if current_user.role == UserRole.STUDENT:
attendance = attendance_service.get_by_student(db, student_id=current_user.id)
elif current_user.role == UserRole.PARENT:
if not student_id:
raise HTTPException(status_code=400, detail="Parent must specify student_id")
attendance = attendance_service.get_by_student(db, student_id=student_id)
elif student_id and date:
attendance = [attendance_service.get_by_student_and_date(db, student_id=student_id, date=date)]
elif student_id:
attendance = attendance_service.get_by_student(db, student_id=student_id)
elif class_id:
attendance = attendance_service.get_by_class(db, class_id=class_id)
elif date:
attendance = attendance_service.get_by_date(db, date=date)
else:
attendance = attendance_service.get_multi(db, skip=skip, limit=limit)
return [a for a in attendance if a is not None]
@router.post("/", response_model=Attendance)
def create_attendance(
*,
db: Session = Depends(deps.get_db),
attendance_in: AttendanceCreate,
current_user: User = Depends(deps.get_current_teacher_or_admin),
) -> Any:
attendance = attendance_service.create_with_teacher(
db, obj_in=attendance_in, teacher_id=current_user.id
)
return attendance
@router.put("/{attendance_id}", response_model=Attendance)
def update_attendance(
*,
db: Session = Depends(deps.get_db),
attendance_id: int,
attendance_in: AttendanceUpdate,
current_user: User = Depends(deps.get_current_teacher_or_admin),
) -> Any:
attendance = attendance_service.get(db, id=attendance_id)
if not attendance:
raise HTTPException(status_code=404, detail="Attendance not found")
attendance = attendance_service.update(db, db_obj=attendance, obj_in=attendance_in)
return attendance
@router.get("/{attendance_id}", response_model=Attendance)
def read_attendance_record(
*,
db: Session = Depends(deps.get_db),
attendance_id: int,
current_user: User = Depends(deps.get_current_active_user),
) -> Any:
attendance = attendance_service.get(db, id=attendance_id)
if not attendance:
raise HTTPException(status_code=404, detail="Attendance not found")
if current_user.role == UserRole.STUDENT and attendance.student_id != current_user.id:
raise HTTPException(status_code=403, detail="Not enough permissions")
elif current_user.role == UserRole.PARENT and attendance.student.parent_id != current_user.id:
raise HTTPException(status_code=403, detail="Not enough permissions")
return attendance

View File

@ -0,0 +1,60 @@
from datetime import timedelta
from typing import Any
from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm
from sqlalchemy.orm import Session
from app.api import deps
from app.core import security
from app.core.config import settings
from app.schemas.token import Token
from app.schemas.user import User, UserCreate
from app.services.user import user_service
router = APIRouter()
@router.post("/login", response_model=Token)
def login_for_access_token(
db: Session = Depends(deps.get_db), form_data: OAuth2PasswordRequestForm = Depends()
) -> Any:
user = user_service.authenticate(
db, email=form_data.username, password=form_data.password
)
if not user:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Incorrect email or password",
headers={"WWW-Authenticate": "Bearer"},
)
elif not user_service.is_active(user):
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST, detail="Inactive user"
)
access_token_expires = timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
return {
"access_token": security.create_access_token(
user.id, expires_delta=access_token_expires
),
"token_type": "bearer",
}
@router.post("/register", response_model=User)
def register_user(
*,
db: Session = Depends(deps.get_db),
user_in: UserCreate,
) -> Any:
user = user_service.get_by_email(db, email=user_in.email)
if user:
raise HTTPException(
status_code=400,
detail="The user with this email already exists in the system.",
)
user = user_service.create(db, obj_in=user_in)
return user
@router.get("/me", response_model=User)
def read_users_me(
db: Session = Depends(deps.get_db),
current_user: User = Depends(deps.get_current_active_user),
) -> Any:
return current_user

View File

@ -0,0 +1,68 @@
from typing import Any, List
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from app.api import deps
from app.models.user import User
from app.schemas.class_schema import Class, ClassCreate, ClassUpdate
from app.services.class_service import class_service
router = APIRouter()
@router.get("/", response_model=List[Class])
def read_classes(
db: Session = Depends(deps.get_db),
skip: int = 0,
limit: int = 100,
current_user: User = Depends(deps.get_current_active_user),
) -> Any:
classes = class_service.get_multi(db, skip=skip, limit=limit)
return classes
@router.post("/", response_model=Class)
def create_class(
*,
db: Session = Depends(deps.get_db),
class_in: ClassCreate,
current_user: User = Depends(deps.get_current_admin_user),
) -> Any:
class_obj = class_service.create(db, obj_in=class_in)
return class_obj
@router.put("/{class_id}", response_model=Class)
def update_class(
*,
db: Session = Depends(deps.get_db),
class_id: int,
class_in: ClassUpdate,
current_user: User = Depends(deps.get_current_admin_user),
) -> Any:
class_obj = class_service.get(db, id=class_id)
if not class_obj:
raise HTTPException(status_code=404, detail="Class not found")
class_obj = class_service.update(db, db_obj=class_obj, obj_in=class_in)
return class_obj
@router.get("/{class_id}", response_model=Class)
def read_class(
*,
db: Session = Depends(deps.get_db),
class_id: int,
current_user: User = Depends(deps.get_current_active_user),
) -> Any:
class_obj = class_service.get(db, id=class_id)
if not class_obj:
raise HTTPException(status_code=404, detail="Class not found")
return class_obj
@router.delete("/{class_id}", response_model=Class)
def delete_class(
*,
db: Session = Depends(deps.get_db),
class_id: int,
current_user: User = Depends(deps.get_current_admin_user),
) -> Any:
class_obj = class_service.get(db, id=class_id)
if not class_obj:
raise HTTPException(status_code=404, detail="Class not found")
class_obj = class_service.remove(db, id=class_id)
return class_obj

View File

@ -0,0 +1,90 @@
from typing import Any, List
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from app.api import deps
from app.models.user import User, UserRole
from app.schemas.grade import Grade, GradeCreate, GradeUpdate
from app.services.grade import grade_service
router = APIRouter()
@router.get("/", response_model=List[Grade])
def read_grades(
db: Session = Depends(deps.get_db),
skip: int = 0,
limit: int = 100,
student_id: int = None,
subject_id: int = None,
current_user: User = Depends(deps.get_current_active_user),
) -> Any:
if current_user.role == UserRole.STUDENT:
grades = grade_service.get_by_student(db, student_id=current_user.id)
elif current_user.role == UserRole.PARENT:
if not student_id:
raise HTTPException(status_code=400, detail="Parent must specify student_id")
grades = grade_service.get_by_student(db, student_id=student_id)
elif student_id and subject_id:
grades = grade_service.get_by_student_and_subject(db, student_id=student_id, subject_id=subject_id)
elif student_id:
grades = grade_service.get_by_student(db, student_id=student_id)
elif subject_id:
grades = grade_service.get_by_subject(db, subject_id=subject_id)
else:
grades = grade_service.get_multi(db, skip=skip, limit=limit)
return grades
@router.post("/", response_model=Grade)
def create_grade(
*,
db: Session = Depends(deps.get_db),
grade_in: GradeCreate,
current_user: User = Depends(deps.get_current_teacher_or_admin),
) -> Any:
grade = grade_service.create(db, obj_in=grade_in)
return grade
@router.put("/{grade_id}", response_model=Grade)
def update_grade(
*,
db: Session = Depends(deps.get_db),
grade_id: int,
grade_in: GradeUpdate,
current_user: User = Depends(deps.get_current_teacher_or_admin),
) -> Any:
grade = grade_service.get(db, id=grade_id)
if not grade:
raise HTTPException(status_code=404, detail="Grade not found")
grade = grade_service.update(db, db_obj=grade, obj_in=grade_in)
return grade
@router.get("/{grade_id}", response_model=Grade)
def read_grade(
*,
db: Session = Depends(deps.get_db),
grade_id: int,
current_user: User = Depends(deps.get_current_active_user),
) -> Any:
grade = grade_service.get(db, id=grade_id)
if not grade:
raise HTTPException(status_code=404, detail="Grade not found")
if current_user.role == UserRole.STUDENT and grade.student_id != current_user.id:
raise HTTPException(status_code=403, detail="Not enough permissions")
elif current_user.role == UserRole.PARENT and grade.student.parent_id != current_user.id:
raise HTTPException(status_code=403, detail="Not enough permissions")
return grade
@router.delete("/{grade_id}", response_model=Grade)
def delete_grade(
*,
db: Session = Depends(deps.get_db),
grade_id: int,
current_user: User = Depends(deps.get_current_teacher_or_admin),
) -> Any:
grade = grade_service.get(db, id=grade_id)
if not grade:
raise HTTPException(status_code=404, detail="Grade not found")
grade = grade_service.remove(db, id=grade_id)
return grade

View File

@ -0,0 +1,82 @@
from typing import Any, List
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from app.api import deps
from app.models.user import User, UserRole
from app.schemas.notification import Notification, NotificationCreate, NotificationUpdate
from app.services.notification import notification_service
router = APIRouter()
@router.get("/", response_model=List[Notification])
def read_notifications(
db: Session = Depends(deps.get_db),
skip: int = 0,
limit: int = 100,
current_user: User = Depends(deps.get_current_active_user),
) -> Any:
if current_user.role in [UserRole.STUDENT, UserRole.PARENT]:
notifications = notification_service.get_by_recipient(db, recipient_id=current_user.id)
else:
notifications = notification_service.get_multi(db, skip=skip, limit=limit)
return notifications
@router.post("/", response_model=Notification)
def create_notification(
*,
db: Session = Depends(deps.get_db),
notification_in: NotificationCreate,
current_user: User = Depends(deps.get_current_teacher_or_admin),
) -> Any:
notification = notification_service.create_with_recipients(
db, obj_in=notification_in, sender_id=current_user.id
)
return notification
@router.put("/{notification_id}", response_model=Notification)
def update_notification(
*,
db: Session = Depends(deps.get_db),
notification_id: int,
notification_in: NotificationUpdate,
current_user: User = Depends(deps.get_current_teacher_or_admin),
) -> Any:
notification = notification_service.get(db, id=notification_id)
if not notification:
raise HTTPException(status_code=404, detail="Notification not found")
if notification.sender_id != current_user.id and current_user.role != UserRole.ADMIN:
raise HTTPException(status_code=403, detail="Not enough permissions")
notification = notification_service.update(db, db_obj=notification, obj_in=notification_in)
return notification
@router.get("/{notification_id}", response_model=Notification)
def read_notification(
*,
db: Session = Depends(deps.get_db),
notification_id: int,
current_user: User = Depends(deps.get_current_active_user),
) -> Any:
notification = notification_service.get(db, id=notification_id)
if not notification:
raise HTTPException(status_code=404, detail="Notification not found")
return notification
@router.delete("/{notification_id}", response_model=Notification)
def delete_notification(
*,
db: Session = Depends(deps.get_db),
notification_id: int,
current_user: User = Depends(deps.get_current_teacher_or_admin),
) -> Any:
notification = notification_service.get(db, id=notification_id)
if not notification:
raise HTTPException(status_code=404, detail="Notification not found")
if notification.sender_id != current_user.id and current_user.role != UserRole.ADMIN:
raise HTTPException(status_code=403, detail="Not enough permissions")
notification = notification_service.remove(db, id=notification_id)
return notification

View File

@ -0,0 +1,68 @@
from typing import Any, List
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from app.api import deps
from app.models.user import User
from app.schemas.subject import Subject, SubjectCreate, SubjectUpdate
from app.services.subject import subject_service
router = APIRouter()
@router.get("/", response_model=List[Subject])
def read_subjects(
db: Session = Depends(deps.get_db),
skip: int = 0,
limit: int = 100,
current_user: User = Depends(deps.get_current_active_user),
) -> Any:
subjects = subject_service.get_multi(db, skip=skip, limit=limit)
return subjects
@router.post("/", response_model=Subject)
def create_subject(
*,
db: Session = Depends(deps.get_db),
subject_in: SubjectCreate,
current_user: User = Depends(deps.get_current_admin_user),
) -> Any:
subject = subject_service.create(db, obj_in=subject_in)
return subject
@router.put("/{subject_id}", response_model=Subject)
def update_subject(
*,
db: Session = Depends(deps.get_db),
subject_id: int,
subject_in: SubjectUpdate,
current_user: User = Depends(deps.get_current_admin_user),
) -> Any:
subject = subject_service.get(db, id=subject_id)
if not subject:
raise HTTPException(status_code=404, detail="Subject not found")
subject = subject_service.update(db, db_obj=subject, obj_in=subject_in)
return subject
@router.get("/{subject_id}", response_model=Subject)
def read_subject(
*,
db: Session = Depends(deps.get_db),
subject_id: int,
current_user: User = Depends(deps.get_current_active_user),
) -> Any:
subject = subject_service.get(db, id=subject_id)
if not subject:
raise HTTPException(status_code=404, detail="Subject not found")
return subject
@router.delete("/{subject_id}", response_model=Subject)
def delete_subject(
*,
db: Session = Depends(deps.get_db),
subject_id: int,
current_user: User = Depends(deps.get_current_admin_user),
) -> Any:
subject = subject_service.get(db, id=subject_id)
if not subject:
raise HTTPException(status_code=404, detail="Subject not found")
subject = subject_service.remove(db, id=subject_id)
return subject

View File

@ -0,0 +1,115 @@
from typing import Any, List
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from app.api import deps
from app.models.user import User, UserRole
from app.schemas.user import User as UserSchema, UserCreate, UserUpdate
from app.services.user import user_service
router = APIRouter()
@router.get("/", response_model=List[UserSchema])
def read_users(
db: Session = Depends(deps.get_db),
skip: int = 0,
limit: int = 100,
current_user: User = Depends(deps.get_current_admin_user),
) -> Any:
users = user_service.get_multi(db, skip=skip, limit=limit)
return users
@router.post("/", response_model=UserSchema)
def create_user(
*,
db: Session = Depends(deps.get_db),
user_in: UserCreate,
current_user: User = Depends(deps.get_current_admin_user),
) -> Any:
user = user_service.get_by_email(db, email=user_in.email)
if user:
raise HTTPException(
status_code=400,
detail="The user with this email already exists in the system",
)
user = user_service.create(db, obj_in=user_in)
return user
@router.put("/{user_id}", response_model=UserSchema)
def update_user(
*,
db: Session = Depends(deps.get_db),
user_id: int,
user_in: UserUpdate,
current_user: User = Depends(deps.get_current_teacher_or_admin),
) -> Any:
user = user_service.get(db, id=user_id)
if not user:
raise HTTPException(
status_code=404,
detail="The user with this id does not exist in the system",
)
if current_user.role == UserRole.TEACHER and user_in.role and user_in.role != UserRole.STUDENT:
raise HTTPException(
status_code=403,
detail="Teachers can only modify student profiles"
)
user = user_service.update(db, db_obj=user, obj_in=user_in)
return user
@router.get("/{user_id}", response_model=UserSchema)
def read_user(
*,
db: Session = Depends(deps.get_db),
user_id: int,
current_user: User = Depends(deps.get_current_active_user),
) -> Any:
user = user_service.get(db, id=user_id)
if not user:
raise HTTPException(status_code=404, detail="User not found")
if current_user.role not in [UserRole.ADMIN, UserRole.TEACHER] and current_user.id != user_id:
raise HTTPException(status_code=403, detail="Not enough permissions")
return user
@router.delete("/{user_id}", response_model=UserSchema)
def delete_user(
*,
db: Session = Depends(deps.get_db),
user_id: int,
current_user: User = Depends(deps.get_current_teacher_or_admin),
) -> Any:
user = user_service.get(db, id=user_id)
if not user:
raise HTTPException(status_code=404, detail="User not found")
if current_user.role == UserRole.TEACHER and user.role != UserRole.STUDENT:
raise HTTPException(
status_code=403,
detail="Teachers can only delete student profiles"
)
user = user_service.remove(db, id=user_id)
return user
@router.get("/students/", response_model=List[UserSchema])
def read_students(
db: Session = Depends(deps.get_db),
skip: int = 0,
limit: int = 100,
current_user: User = Depends(deps.get_current_teacher_or_admin),
) -> Any:
students = user_service.get_students(db, skip=skip, limit=limit)
return students
@router.get("/teachers/", response_model=List[UserSchema])
def read_teachers(
db: Session = Depends(deps.get_db),
skip: int = 0,
limit: int = 100,
current_user: User = Depends(deps.get_current_admin_user),
) -> Any:
teachers = user_service.get_teachers(db, skip=skip, limit=limit)
return teachers

0
app/core/__init__.py Normal file
View File

20
app/core/config.py Normal file
View File

@ -0,0 +1,20 @@
import os
from pydantic import BaseSettings
class Settings(BaseSettings):
PROJECT_NAME: str = "School Portal API"
VERSION: str = "1.0.0"
API_V1_STR: str = "/api/v1"
SECRET_KEY: str = os.getenv("SECRET_KEY", "your-secret-key-change-in-production")
ACCESS_TOKEN_EXPIRE_MINUTES: int = 60 * 24 * 8 # 8 days
DATABASE_URL: str = os.getenv("DATABASE_URL", "sqlite:////app/storage/db/db.sqlite")
FIRST_SUPERUSER_EMAIL: str = os.getenv("FIRST_SUPERUSER_EMAIL", "admin@schoolportal.com")
FIRST_SUPERUSER_PASSWORD: str = os.getenv("FIRST_SUPERUSER_PASSWORD", "admin123")
class Config:
case_sensitive = True
settings = Settings()

38
app/core/security.py Normal file
View File

@ -0,0 +1,38 @@
from datetime import datetime, timedelta
from typing import Optional, Union
from jose import JWTError, jwt
from passlib.context import CryptContext
from app.core.config import settings
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
ALGORITHM = "HS256"
def create_access_token(
subject: Union[str, int], expires_delta: Optional[timedelta] = None
):
if expires_delta:
expire = datetime.utcnow() + expires_delta
else:
expire = datetime.utcnow() + timedelta(
minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES
)
to_encode = {"exp": expire, "sub": str(subject)}
encoded_jwt = jwt.encode(to_encode, settings.SECRET_KEY, algorithm=ALGORITHM)
return encoded_jwt
def verify_password(plain_password: str, hashed_password: str) -> bool:
return pwd_context.verify(plain_password, hashed_password)
def get_password_hash(password: str) -> str:
return pwd_context.hash(password)
def verify_token(token: str) -> Optional[str]:
try:
payload = jwt.decode(token, settings.SECRET_KEY, algorithms=[ALGORITHM])
user_id: str = payload.get("sub")
if user_id is None:
return None
return user_id
except JWTError:
return None

0
app/db/__init__.py Normal file
View File

3
app/db/base.py Normal file
View File

@ -0,0 +1,3 @@
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()

15
app/db/session.py Normal file
View File

@ -0,0 +1,15 @@
from pathlib import Path
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
DB_DIR = Path("/app/storage/db")
DB_DIR.mkdir(parents=True, exist_ok=True)
SQLALCHEMY_DATABASE_URL = f"sqlite:///{DB_DIR}/db.sqlite"
engine = create_engine(
SQLALCHEMY_DATABASE_URL,
connect_args={"check_same_thread": False}
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

8
app/models/__init__.py Normal file
View File

@ -0,0 +1,8 @@
from .user import User
from .class_model import Class
from .subject import Subject
from .grade import Grade
from .attendance import Attendance
from .notification import Notification
__all__ = ["User", "Class", "Subject", "Grade", "Attendance", "Notification"]

21
app/models/attendance.py Normal file
View File

@ -0,0 +1,21 @@
from sqlalchemy import Column, Integer, String, DateTime, ForeignKey, Date, Boolean
from sqlalchemy.orm import relationship
from sqlalchemy.sql import func
from app.db.base import Base
class Attendance(Base):
__tablename__ = "attendance"
id = Column(Integer, primary_key=True, index=True)
student_id = Column(Integer, ForeignKey("users.id"), nullable=False)
class_id = Column(Integer, ForeignKey("classes.id"), nullable=False)
date = Column(Date, nullable=False)
is_present = Column(Boolean, default=False)
remarks = Column(String)
marked_by = Column(Integer, ForeignKey("users.id"), nullable=False)
created_at = Column(DateTime(timezone=True), server_default=func.now())
updated_at = Column(DateTime(timezone=True), onupdate=func.now())
student = relationship("User", foreign_keys=[student_id], back_populates="attendance_records")
class_ref = relationship("Class", back_populates="attendance_records")
teacher = relationship("User", foreign_keys=[marked_by])

26
app/models/class_model.py Normal file
View File

@ -0,0 +1,26 @@
from sqlalchemy import Column, Integer, String, DateTime, Table, ForeignKey
from sqlalchemy.orm import relationship
from sqlalchemy.sql import func
from app.db.base import Base
teacher_classes = Table(
"teacher_classes",
Base.metadata,
Column("teacher_id", Integer, ForeignKey("users.id"), primary_key=True),
Column("class_id", Integer, ForeignKey("classes.id"), primary_key=True),
)
class Class(Base):
__tablename__ = "classes"
id = Column(Integer, primary_key=True, index=True)
name = Column(String, nullable=False)
grade_level = Column(String, nullable=False)
academic_year = Column(String, nullable=False)
created_at = Column(DateTime(timezone=True), server_default=func.now())
updated_at = Column(DateTime(timezone=True), onupdate=func.now())
students = relationship("User", back_populates="assigned_class")
teachers = relationship("User", secondary=teacher_classes, back_populates="taught_classes")
subjects = relationship("Subject", back_populates="class_ref")
attendance_records = relationship("Attendance", back_populates="class_ref")

21
app/models/grade.py Normal file
View File

@ -0,0 +1,21 @@
from sqlalchemy import Column, Integer, String, DateTime, ForeignKey, Float
from sqlalchemy.orm import relationship
from sqlalchemy.sql import func
from app.db.base import Base
class Grade(Base):
__tablename__ = "grades"
id = Column(Integer, primary_key=True, index=True)
student_id = Column(Integer, ForeignKey("users.id"), nullable=False)
subject_id = Column(Integer, ForeignKey("subjects.id"), nullable=False)
score = Column(Float, nullable=False)
max_score = Column(Float, nullable=False, default=100.0)
grade_type = Column(String, nullable=False) # quiz, exam, assignment, etc.
description = Column(String)
graded_at = Column(DateTime(timezone=True), server_default=func.now())
created_at = Column(DateTime(timezone=True), server_default=func.now())
updated_at = Column(DateTime(timezone=True), onupdate=func.now())
student = relationship("User", back_populates="grades")
subject = relationship("Subject", back_populates="grades")

View File

@ -0,0 +1,28 @@
from sqlalchemy import Column, Integer, String, DateTime, ForeignKey, Boolean, Text, Table
from sqlalchemy.orm import relationship
from sqlalchemy.sql import func
from app.db.base import Base
notification_recipients = Table(
"notification_recipients",
Base.metadata,
Column("notification_id", Integer, ForeignKey("notifications.id"), primary_key=True),
Column("user_id", Integer, ForeignKey("users.id"), primary_key=True),
Column("is_read", Boolean, default=False),
Column("read_at", DateTime(timezone=True))
)
class Notification(Base):
__tablename__ = "notifications"
id = Column(Integer, primary_key=True, index=True)
title = Column(String, nullable=False)
message = Column(Text, nullable=False)
sender_id = Column(Integer, ForeignKey("users.id"), nullable=False)
notification_type = Column(String, nullable=False) # announcement, message, alert
priority = Column(String, default="normal") # low, normal, high, urgent
created_at = Column(DateTime(timezone=True), server_default=func.now())
updated_at = Column(DateTime(timezone=True), onupdate=func.now())
sender = relationship("User", foreign_keys=[sender_id], back_populates="sent_notifications")
recipients = relationship("User", secondary=notification_recipients, back_populates="received_notifications")

20
app/models/subject.py Normal file
View File

@ -0,0 +1,20 @@
from sqlalchemy import Column, Integer, String, DateTime, ForeignKey
from sqlalchemy.orm import relationship
from sqlalchemy.sql import func
from app.db.base import Base
class Subject(Base):
__tablename__ = "subjects"
id = Column(Integer, primary_key=True, index=True)
name = Column(String, nullable=False)
code = Column(String, unique=True, nullable=False)
description = Column(String)
class_id = Column(Integer, ForeignKey("classes.id"), nullable=False)
teacher_id = Column(Integer, ForeignKey("users.id"), nullable=False)
created_at = Column(DateTime(timezone=True), server_default=func.now())
updated_at = Column(DateTime(timezone=True), onupdate=func.now())
class_ref = relationship("Class", back_populates="subjects")
teacher = relationship("User")
grades = relationship("Grade", back_populates="subject")

35
app/models/user.py Normal file
View File

@ -0,0 +1,35 @@
from sqlalchemy import Column, Integer, String, Boolean, DateTime, Enum, ForeignKey
from sqlalchemy.orm import relationship
from sqlalchemy.sql import func
import enum
from app.db.base import Base
class UserRole(str, enum.Enum):
ADMIN = "admin"
TEACHER = "teacher"
STUDENT = "student"
PARENT = "parent"
class User(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True, index=True)
email = Column(String, unique=True, index=True, nullable=False)
hashed_password = Column(String, nullable=False)
first_name = Column(String, nullable=False)
last_name = Column(String, nullable=False)
role = Column(Enum(UserRole), nullable=False)
is_active = Column(Boolean, default=True)
created_at = Column(DateTime(timezone=True), server_default=func.now())
updated_at = Column(DateTime(timezone=True), onupdate=func.now())
parent_id = Column(Integer, ForeignKey("users.id"), nullable=True)
class_id = Column(Integer, ForeignKey("classes.id"), nullable=True)
parent = relationship("User", remote_side=[id], backref="children")
assigned_class = relationship("Class", back_populates="students")
taught_classes = relationship("Class", secondary="teacher_classes", back_populates="teachers")
grades = relationship("Grade", back_populates="student")
attendance_records = relationship("Attendance", back_populates="student")
sent_notifications = relationship("Notification", foreign_keys="Notification.sender_id", back_populates="sender")
received_notifications = relationship("Notification", secondary="notification_recipients", back_populates="recipients")

17
app/schemas/__init__.py Normal file
View File

@ -0,0 +1,17 @@
from .user import User, UserCreate, UserUpdate, UserInDB
from .class_schema import Class, ClassCreate, ClassUpdate
from .subject import Subject, SubjectCreate, SubjectUpdate
from .grade import Grade, GradeCreate, GradeUpdate
from .attendance import Attendance, AttendanceCreate, AttendanceUpdate
from .notification import Notification, NotificationCreate, NotificationUpdate
from .token import Token, TokenPayload
__all__ = [
"User", "UserCreate", "UserUpdate", "UserInDB",
"Class", "ClassCreate", "ClassUpdate",
"Subject", "SubjectCreate", "SubjectUpdate",
"Grade", "GradeCreate", "GradeUpdate",
"Attendance", "AttendanceCreate", "AttendanceUpdate",
"Notification", "NotificationCreate", "NotificationUpdate",
"Token", "TokenPayload"
]

29
app/schemas/attendance.py Normal file
View File

@ -0,0 +1,29 @@
from typing import Optional
from pydantic import BaseModel
from datetime import datetime, date
class AttendanceBase(BaseModel):
student_id: int
class_id: int
date: date
is_present: bool = False
remarks: Optional[str] = None
class AttendanceCreate(AttendanceBase):
pass
class AttendanceUpdate(BaseModel):
is_present: Optional[bool] = None
remarks: Optional[str] = None
class AttendanceInDBBase(AttendanceBase):
id: int
marked_by: int
created_at: datetime
updated_at: Optional[datetime] = None
class Config:
orm_mode = True
class Attendance(AttendanceInDBBase):
pass

View File

@ -0,0 +1,27 @@
from typing import Optional
from pydantic import BaseModel
from datetime import datetime
class ClassBase(BaseModel):
name: str
grade_level: str
academic_year: str
class ClassCreate(ClassBase):
pass
class ClassUpdate(BaseModel):
name: Optional[str] = None
grade_level: Optional[str] = None
academic_year: Optional[str] = None
class ClassInDBBase(ClassBase):
id: int
created_at: datetime
updated_at: Optional[datetime] = None
class Config:
orm_mode = True
class Class(ClassInDBBase):
pass

32
app/schemas/grade.py Normal file
View File

@ -0,0 +1,32 @@
from typing import Optional
from pydantic import BaseModel
from datetime import datetime
class GradeBase(BaseModel):
student_id: int
subject_id: int
score: float
max_score: float = 100.0
grade_type: str
description: Optional[str] = None
class GradeCreate(GradeBase):
pass
class GradeUpdate(BaseModel):
score: Optional[float] = None
max_score: Optional[float] = None
grade_type: Optional[str] = None
description: Optional[str] = None
class GradeInDBBase(GradeBase):
id: int
graded_at: datetime
created_at: datetime
updated_at: Optional[datetime] = None
class Config:
orm_mode = True
class Grade(GradeInDBBase):
pass

View File

@ -0,0 +1,30 @@
from typing import Optional, List
from pydantic import BaseModel
from datetime import datetime
class NotificationBase(BaseModel):
title: str
message: str
notification_type: str
priority: str = "normal"
class NotificationCreate(NotificationBase):
recipient_ids: List[int]
class NotificationUpdate(BaseModel):
title: Optional[str] = None
message: Optional[str] = None
notification_type: Optional[str] = None
priority: Optional[str] = None
class NotificationInDBBase(NotificationBase):
id: int
sender_id: int
created_at: datetime
updated_at: Optional[datetime] = None
class Config:
orm_mode = True
class Notification(NotificationInDBBase):
pass

31
app/schemas/subject.py Normal file
View File

@ -0,0 +1,31 @@
from typing import Optional
from pydantic import BaseModel
from datetime import datetime
class SubjectBase(BaseModel):
name: str
code: str
description: Optional[str] = None
class_id: int
teacher_id: int
class SubjectCreate(SubjectBase):
pass
class SubjectUpdate(BaseModel):
name: Optional[str] = None
code: Optional[str] = None
description: Optional[str] = None
class_id: Optional[int] = None
teacher_id: Optional[int] = None
class SubjectInDBBase(SubjectBase):
id: int
created_at: datetime
updated_at: Optional[datetime] = None
class Config:
orm_mode = True
class Subject(SubjectInDBBase):
pass

9
app/schemas/token.py Normal file
View File

@ -0,0 +1,9 @@
from typing import Optional
from pydantic import BaseModel
class Token(BaseModel):
access_token: str
token_type: str
class TokenPayload(BaseModel):
sub: Optional[int] = None

40
app/schemas/user.py Normal file
View File

@ -0,0 +1,40 @@
from typing import Optional
from pydantic import BaseModel, EmailStr
from datetime import datetime
from app.models.user import UserRole
class UserBase(BaseModel):
email: EmailStr
first_name: str
last_name: str
role: UserRole
is_active: bool = True
parent_id: Optional[int] = None
class_id: Optional[int] = None
class UserCreate(UserBase):
password: str
class UserUpdate(BaseModel):
email: Optional[EmailStr] = None
first_name: Optional[str] = None
last_name: Optional[str] = None
role: Optional[UserRole] = None
is_active: Optional[bool] = None
parent_id: Optional[int] = None
class_id: Optional[int] = None
password: Optional[str] = None
class UserInDBBase(UserBase):
id: int
created_at: datetime
updated_at: Optional[datetime] = None
class Config:
orm_mode = True
class User(UserInDBBase):
pass
class UserInDB(UserInDBBase):
hashed_password: str

0
app/services/__init__.py Normal file
View File

View File

@ -0,0 +1,33 @@
from typing import List
from sqlalchemy.orm import Session
from datetime import date
from app.models.attendance import Attendance
from app.schemas.attendance import AttendanceCreate, AttendanceUpdate
from app.services.base import CRUDBase
class CRUDAttendance(CRUDBase[Attendance, AttendanceCreate, AttendanceUpdate]):
def get_by_student(self, db: Session, *, student_id: int) -> List[Attendance]:
return db.query(Attendance).filter(Attendance.student_id == student_id).all()
def get_by_class(self, db: Session, *, class_id: int) -> List[Attendance]:
return db.query(Attendance).filter(Attendance.class_id == class_id).all()
def get_by_date(self, db: Session, *, date: date) -> List[Attendance]:
return db.query(Attendance).filter(Attendance.date == date).all()
def get_by_student_and_date(self, db: Session, *, student_id: int, date: date) -> Attendance:
return db.query(Attendance).filter(
Attendance.student_id == student_id,
Attendance.date == date
).first()
def create_with_teacher(self, db: Session, *, obj_in: AttendanceCreate, teacher_id: int) -> Attendance:
obj_in_data = obj_in.dict()
obj_in_data["marked_by"] = teacher_id
db_obj = Attendance(**obj_in_data)
db.add(db_obj)
db.commit()
db.refresh(db_obj)
return db_obj
attendance_service = CRUDAttendance(Attendance)

55
app/services/base.py Normal file
View File

@ -0,0 +1,55 @@
from typing import Any, Dict, Generic, List, Optional, Type, TypeVar, Union
from fastapi.encoders import jsonable_encoder
from pydantic import BaseModel
from sqlalchemy.orm import Session
from app.db.base import Base
ModelType = TypeVar("ModelType", bound=Base)
CreateSchemaType = TypeVar("CreateSchemaType", bound=BaseModel)
UpdateSchemaType = TypeVar("UpdateSchemaType", bound=BaseModel)
class CRUDBase(Generic[ModelType, CreateSchemaType, UpdateSchemaType]):
def __init__(self, model: Type[ModelType]):
self.model = model
def get(self, db: Session, id: Any) -> Optional[ModelType]:
return db.query(self.model).filter(self.model.id == id).first()
def get_multi(
self, db: Session, *, skip: int = 0, limit: int = 100
) -> List[ModelType]:
return db.query(self.model).offset(skip).limit(limit).all()
def create(self, db: Session, *, obj_in: CreateSchemaType) -> ModelType:
obj_in_data = jsonable_encoder(obj_in)
db_obj = self.model(**obj_in_data)
db.add(db_obj)
db.commit()
db.refresh(db_obj)
return db_obj
def update(
self,
db: Session,
*,
db_obj: ModelType,
obj_in: Union[UpdateSchemaType, Dict[str, Any]]
) -> ModelType:
obj_data = jsonable_encoder(db_obj)
if isinstance(obj_in, dict):
update_data = obj_in
else:
update_data = obj_in.dict(exclude_unset=True)
for field in obj_data:
if field in update_data:
setattr(db_obj, field, update_data[field])
db.add(db_obj)
db.commit()
db.refresh(db_obj)
return db_obj
def remove(self, db: Session, *, id: int) -> ModelType:
obj = db.query(self.model).get(id)
db.delete(obj)
db.commit()
return obj

View File

@ -0,0 +1,14 @@
from typing import List
from sqlalchemy.orm import Session
from app.models.class_model import Class
from app.schemas.class_schema import ClassCreate, ClassUpdate
from app.services.base import CRUDBase
class CRUDClass(CRUDBase[Class, ClassCreate, ClassUpdate]):
def get_by_grade_level(self, db: Session, *, grade_level: str) -> List[Class]:
return db.query(Class).filter(Class.grade_level == grade_level).all()
def get_by_academic_year(self, db: Session, *, academic_year: str) -> List[Class]:
return db.query(Class).filter(Class.academic_year == academic_year).all()
class_service = CRUDClass(Class)

20
app/services/grade.py Normal file
View File

@ -0,0 +1,20 @@
from typing import List
from sqlalchemy.orm import Session
from app.models.grade import Grade
from app.schemas.grade import GradeCreate, GradeUpdate
from app.services.base import CRUDBase
class CRUDGrade(CRUDBase[Grade, GradeCreate, GradeUpdate]):
def get_by_student(self, db: Session, *, student_id: int) -> List[Grade]:
return db.query(Grade).filter(Grade.student_id == student_id).all()
def get_by_subject(self, db: Session, *, subject_id: int) -> List[Grade]:
return db.query(Grade).filter(Grade.subject_id == subject_id).all()
def get_by_student_and_subject(self, db: Session, *, student_id: int, subject_id: int) -> List[Grade]:
return db.query(Grade).filter(
Grade.student_id == student_id,
Grade.subject_id == subject_id
).all()
grade_service = CRUDGrade(Grade)

View File

@ -0,0 +1,35 @@
from typing import List
from sqlalchemy.orm import Session
from app.models.notification import Notification
from app.schemas.notification import NotificationCreate, NotificationUpdate
from app.services.base import CRUDBase
class CRUDNotification(CRUDBase[Notification, NotificationCreate, NotificationUpdate]):
def create_with_recipients(self, db: Session, *, obj_in: NotificationCreate, sender_id: int) -> Notification:
recipient_ids = obj_in.recipient_ids
obj_in_data = obj_in.dict(exclude={"recipient_ids"})
obj_in_data["sender_id"] = sender_id
db_obj = Notification(**obj_in_data)
db.add(db_obj)
db.flush()
for recipient_id in recipient_ids:
db.execute(
"INSERT INTO notification_recipients (notification_id, user_id) VALUES (?, ?)",
(db_obj.id, recipient_id)
)
db.commit()
db.refresh(db_obj)
return db_obj
def get_by_sender(self, db: Session, *, sender_id: int) -> List[Notification]:
return db.query(Notification).filter(Notification.sender_id == sender_id).all()
def get_by_recipient(self, db: Session, *, recipient_id: int) -> List[Notification]:
return db.query(Notification).join(
Notification.recipients
).filter_by(id=recipient_id).all()
notification_service = CRUDNotification(Notification)

14
app/services/subject.py Normal file
View File

@ -0,0 +1,14 @@
from typing import List
from sqlalchemy.orm import Session
from app.models.subject import Subject
from app.schemas.subject import SubjectCreate, SubjectUpdate
from app.services.base import CRUDBase
class CRUDSubject(CRUDBase[Subject, SubjectCreate, SubjectUpdate]):
def get_by_class(self, db: Session, *, class_id: int) -> List[Subject]:
return db.query(Subject).filter(Subject.class_id == class_id).all()
def get_by_teacher(self, db: Session, *, teacher_id: int) -> List[Subject]:
return db.query(Subject).filter(Subject.teacher_id == teacher_id).all()
subject_service = CRUDSubject(Subject)

63
app/services/user.py Normal file
View File

@ -0,0 +1,63 @@
from typing import Any, Dict, Optional, Union, List
from sqlalchemy.orm import Session
from app.core.security import get_password_hash, verify_password
from app.models.user import User, UserRole
from app.schemas.user import UserCreate, UserUpdate
from app.services.base import CRUDBase
class CRUDUser(CRUDBase[User, UserCreate, UserUpdate]):
def get_by_email(self, db: Session, *, email: str) -> Optional[User]:
return db.query(User).filter(User.email == email).first()
def create(self, db: Session, *, obj_in: UserCreate) -> User:
db_obj = User(
email=obj_in.email,
hashed_password=get_password_hash(obj_in.password),
first_name=obj_in.first_name,
last_name=obj_in.last_name,
role=obj_in.role,
is_active=obj_in.is_active,
parent_id=obj_in.parent_id,
class_id=obj_in.class_id,
)
db.add(db_obj)
db.commit()
db.refresh(db_obj)
return db_obj
def update(
self, db: Session, *, db_obj: User, obj_in: Union[UserUpdate, Dict[str, Any]]
) -> User:
if isinstance(obj_in, dict):
update_data = obj_in
else:
update_data = obj_in.dict(exclude_unset=True)
if "password" in update_data:
hashed_password = get_password_hash(update_data["password"])
del update_data["password"]
update_data["hashed_password"] = hashed_password
return super().update(db, db_obj=db_obj, obj_in=update_data)
def authenticate(self, db: Session, *, email: str, password: str) -> Optional[User]:
user = self.get_by_email(db, email=email)
if not user:
return None
if not verify_password(password, user.hashed_password):
return None
return user
def is_active(self, user: User) -> bool:
return user.is_active
def get_students(self, db: Session, *, skip: int = 0, limit: int = 100) -> List[User]:
return db.query(User).filter(User.role == UserRole.STUDENT).offset(skip).limit(limit).all()
def get_teachers(self, db: Session, *, skip: int = 0, limit: int = 100) -> List[User]:
return db.query(User).filter(User.role == UserRole.TEACHER).offset(skip).limit(limit).all()
def get_parents(self, db: Session, *, skip: int = 0, limit: int = 100) -> List[User]:
return db.query(User).filter(User.role == UserRole.PARENT).offset(skip).limit(limit).all()
user_service = CRUDUser(User)

37
main.py Normal file
View File

@ -0,0 +1,37 @@
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from app.api.v1.api import api_router
from app.core.config import settings
app = FastAPI(
title=settings.PROJECT_NAME,
version=settings.VERSION,
description="School Portal API - A comprehensive school management system",
openapi_url="/openapi.json",
docs_url="/docs",
redoc_url="/redoc"
)
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
app.include_router(api_router, prefix="/api/v1")
@app.get("/")
async def root():
return {
"title": settings.PROJECT_NAME,
"version": settings.VERSION,
"description": "School Portal API - A comprehensive school management system",
"documentation": "/docs",
"health": "/health"
}
@app.get("/health")
async def health_check():
return {"status": "healthy", "service": "School Portal API"}

9
requirements.txt Normal file
View File

@ -0,0 +1,9 @@
fastapi==0.104.1
uvicorn[standard]==0.24.0
sqlalchemy==2.0.23
alembic==1.12.1
pydantic[email]==2.5.0
python-jose[cryptography]==3.3.0
passlib[bcrypt]==1.7.4
python-multipart==0.0.6
ruff==0.1.6