mirror of
https://github.com/jhd3197/ServerKit.git
synced 2026-11-10 05:53:05 +09:00
Juan Denis
08d5c4bade
Add full SSO (Google, GitHub, generic OIDC, SAML) support and integrate Alembic-based DB migrations. Introduces OAuthIdentity model, sso_service, SSO API blueprint (authorize/callback/link/unlink/admin/config/test), and settings to enable/configure providers and enforce SSO-only logins. Integrates Flask-Migrate and a MigrationService with API endpoints for status/backup/apply/history and includes Alembic scaffolding (migrations/versions/001_baseline.py). Update app init to register SSO and migrations blueprints and initialize Migrate; replace fragile auto-column hack with MigrationService.prepare/check. Update User model (nullable password_hash, auth_provider, has_password/to_dict fields), add AuditLog SSO actions, add frontend SSO components/pages and a migration UI, and update ROADMAP and requirements. Notes: token encryption uses Fernet derived from SECRET_KEY; migrations may add the new auth_provider column at startup.
55 lines
1.8 KiB
Python
55 lines
1.8 KiB
Python
from flask import Blueprint, jsonify, current_app
|
|
from flask_jwt_extended import jwt_required, get_jwt_identity
|
|
|
|
from app.models import User
|
|
from app.services.migration_service import MigrationService
|
|
|
|
migrations_bp = Blueprint('migrations', __name__)
|
|
|
|
|
|
@migrations_bp.route('/status', methods=['GET'])
|
|
def get_migration_status():
|
|
"""Check if migrations are pending. No auth required (called before login)."""
|
|
status = MigrationService.get_status()
|
|
return jsonify(status), 200
|
|
|
|
|
|
@migrations_bp.route('/backup', methods=['POST'])
|
|
@jwt_required()
|
|
def create_backup():
|
|
"""Create a database backup before applying migrations. Admin only."""
|
|
user = User.query.get(get_jwt_identity())
|
|
if not user or user.role != User.ROLE_ADMIN:
|
|
return jsonify({'error': 'Admin access required'}), 403
|
|
|
|
result = MigrationService.create_backup(current_app)
|
|
if result['success']:
|
|
return jsonify(result), 200
|
|
return jsonify(result), 500
|
|
|
|
|
|
@migrations_bp.route('/apply', methods=['POST'])
|
|
@jwt_required()
|
|
def apply_migrations():
|
|
"""Apply all pending migrations. Admin only."""
|
|
user = User.query.get(get_jwt_identity())
|
|
if not user or user.role != User.ROLE_ADMIN:
|
|
return jsonify({'error': 'Admin access required'}), 403
|
|
|
|
result = MigrationService.apply_migrations(current_app)
|
|
if result['success']:
|
|
return jsonify(result), 200
|
|
return jsonify(result), 500
|
|
|
|
|
|
@migrations_bp.route('/history', methods=['GET'])
|
|
@jwt_required()
|
|
def get_migration_history():
|
|
"""Return all migration revisions. Admin only."""
|
|
user = User.query.get(get_jwt_identity())
|
|
if not user or user.role != User.ROLE_ADMIN:
|
|
return jsonify({'error': 'Admin access required'}), 403
|
|
|
|
history = MigrationService.get_migration_history(current_app)
|
|
return jsonify({'revisions': history}), 200
|