196 lines
6.9 KiB
Python
196 lines
6.9 KiB
Python
import os
|
|
import sys
|
|
import uuid
|
|
from pathlib import Path
|
|
|
|
# Add project root to path for proper imports
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
|
|
|
from fastapi import FastAPI, Depends, status, HTTPException
|
|
from fastapi.staticfiles import StaticFiles
|
|
from fastapi.responses import HTMLResponse, Response
|
|
from fastapi.templating import Jinja2Templates
|
|
from fastapi.requests import Request
|
|
from fastapi.middleware.cors import CORSMiddleware
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlalchemy import select
|
|
|
|
from auth import auth_backend, fastapi_users, get_user_manager, current_active_user, current_superuser
|
|
from database import engine, Base, get_async_session, AsyncSessionLocal
|
|
from models import User, URLList
|
|
from schemas import UserRead, UserCreate, UserUpdate
|
|
from config import get_settings
|
|
from routers import auth as auth_router
|
|
from routers import users as users_router
|
|
from routers import url_lists as url_lists_router
|
|
|
|
settings = get_settings()
|
|
|
|
app = FastAPI(
|
|
title=settings.APP_NAME,
|
|
description="A web application to manage lists of URLs.",
|
|
version="0.2.0",
|
|
docs_url="/api/docs",
|
|
redoc_url="/api/redoc"
|
|
)
|
|
|
|
# --- CORS Configuration ---
|
|
app.add_middleware(
|
|
CORSMiddleware,
|
|
allow_origins=settings.cors_origins_list,
|
|
allow_credentials=True,
|
|
allow_methods=["*"],
|
|
allow_headers=["*"],
|
|
)
|
|
|
|
# --- Static Files and Templates ---
|
|
BASE_DIR = Path(__file__).resolve().parent
|
|
STATIC_DIR = BASE_DIR / "static"
|
|
TEMPLATES_DIR = STATIC_DIR / "templates"
|
|
|
|
# Ensure directories exist
|
|
STATIC_DIR.mkdir(exist_ok=True)
|
|
TEMPLATES_DIR.mkdir(exist_ok=True)
|
|
|
|
# Mount static files (CSS, JS, etc.)
|
|
app.mount("/static", StaticFiles(directory=STATIC_DIR), name="static")
|
|
|
|
# Setup Jinja2 templates
|
|
templates = Jinja2Templates(directory=str(TEMPLATES_DIR))
|
|
|
|
# Helper function for URL generation (similar to Flask's url_for)
|
|
def url_for(name: str, **params) -> str:
|
|
"""Generate URL paths for templates. Similar to Flask's url_for."""
|
|
routes = {
|
|
"home": "/",
|
|
"login": "/login",
|
|
"register": "/register",
|
|
"dashboard": "/dashboard",
|
|
"create_list": "/dashboard/list/create",
|
|
"list_read": "/list-read/{slug}",
|
|
"logout": "/api/auth/logout",
|
|
"api_docs": "/api/docs",
|
|
"api_redoc": "/api/redoc",
|
|
}
|
|
|
|
static_routes = {
|
|
"css_output": "/static/output.css",
|
|
"js_auth": "/static/js/auth.js",
|
|
"js_api": "/static/js/api.js",
|
|
}
|
|
|
|
# Check if it's a static route
|
|
if name in static_routes:
|
|
return static_routes[name]
|
|
|
|
# Check if it's a regular route
|
|
if name in routes:
|
|
url = routes[name]
|
|
# Replace parameters if provided
|
|
for key, value in params.items():
|
|
url = url.replace(f"{{{key}}}", str(value))
|
|
return url
|
|
|
|
raise ValueError(f"Unknown route: {name}")
|
|
|
|
# Add url_for to all template contexts
|
|
def get_template_context(request: Request, **extra):
|
|
"""Create standard template context with url_for function."""
|
|
context = {
|
|
"request": request,
|
|
"app_name": settings.APP_NAME,
|
|
"url_for": url_for,
|
|
"current_user": None,
|
|
}
|
|
context.update(extra)
|
|
return context
|
|
# --- Database Initialization and Default Admin Creation ---
|
|
@app.on_event("startup")
|
|
async def on_startup():
|
|
# Create database tables if they don't exist
|
|
async with engine.begin() as conn:
|
|
await conn.run_sync(Base.metadata.create_all)
|
|
|
|
# Check for and create default admin user on first run
|
|
async with AsyncSessionLocal() as session:
|
|
from auth import get_user_manager as _get_user_manager
|
|
user_db = SQLAlchemyUserDatabase(session, User)
|
|
user_manager = UserManager(user_db)
|
|
|
|
# Check if any user exists
|
|
result = await session.execute(select(User))
|
|
users = result.scalars().all()
|
|
|
|
if len(users) == 0:
|
|
print("No users found. Creating default admin user...")
|
|
try:
|
|
# Use UserCreate schema to ensure required fields are present
|
|
admin_user_data = UserCreate(
|
|
email="localadmin@example.com",
|
|
password="AdminSuper123!",
|
|
is_superuser=True,
|
|
is_verified=True, # Mark as verified for immediate use
|
|
)
|
|
admin_user = await user_manager.create(
|
|
admin_user_data,
|
|
safe=False # Allow setting is_superuser directly
|
|
)
|
|
print(f"Default admin user '{admin_user.email}' created successfully.")
|
|
except Exception as e:
|
|
print(f"Error creating default admin user: {e}")
|
|
|
|
# --- Import for UserManager after it's defined
|
|
from fastapi_users.db import SQLAlchemyUserDatabase
|
|
from auth import UserManager
|
|
|
|
# --- API Routers ---
|
|
app.include_router(auth_router.router, prefix="/api/auth")
|
|
app.include_router(users_router.router, prefix="/api") # e.g., /api/users
|
|
app.include_router(url_lists_router.router, prefix="/api") # e.g., /api/url-lists
|
|
app.include_router(url_lists_router.public_router, prefix="/api") # e.g., /api/public-lists
|
|
|
|
# --- Public Raw URL List Endpoint ---
|
|
@app.get("/list-read/{unique_slug}", response_class=Response, responses={
|
|
200: {"content": {"text/plain": {}}},
|
|
404: {"description": "URL list not found"}
|
|
})
|
|
async def read_url_list_raw(
|
|
unique_slug: str,
|
|
session: AsyncSession = Depends(get_async_session)
|
|
):
|
|
result = await session.execute(
|
|
select(URLList).filter(URLList.unique_slug == unique_slug)
|
|
)
|
|
url_list = result.scalar_one_or_none()
|
|
if not url_list:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="URL list not found")
|
|
|
|
return Response(content=url_list.urls, media_type="text/plain")
|
|
|
|
# --- Frontend Routes with Jinja2 Templates ---
|
|
@app.get("/", response_class=HTMLResponse)
|
|
async def read_root(request: Request):
|
|
return templates.TemplateResponse("index.html", get_template_context(request))
|
|
|
|
@app.get("/login", response_class=HTMLResponse)
|
|
async def login_page(request: Request):
|
|
return templates.TemplateResponse("login.html", get_template_context(request))
|
|
|
|
@app.get("/register", response_class=HTMLResponse)
|
|
async def register_page(request: Request):
|
|
return templates.TemplateResponse("register.html", get_template_context(request))
|
|
|
|
@app.get("/dashboard", response_class=HTMLResponse)
|
|
async def dashboard_page(request: Request):
|
|
return templates.TemplateResponse("dashboard.html", get_template_context(request))
|
|
|
|
@app.get("/dashboard/list/create", response_class=HTMLResponse)
|
|
async def create_list_page(request: Request):
|
|
return templates.TemplateResponse("create_list.html", get_template_context(request))
|
|
|
|
# Catch-all for any other routes - redirect to home
|
|
@app.get("/{full_path:path}", response_class=HTMLResponse)
|
|
async def catch_all(request: Request, full_path: str):
|
|
# Return home page for any unmatched routes (SPA-like behavior)
|
|
return templates.TemplateResponse("index.html", get_template_context(request))
|