import os
import devplacepy.db_client
import asyncio
import logging
import time
from collections import defaultdict
from contextlib import asynccontextmanager
from pathlib import Path
from fastapi import FastAPI, Request
from fastapi.exceptions import RequestValidationError
from fastapi.responses import HTMLResponse, JSONResponse, RedirectResponse
from fastapi.staticfiles import StaticFiles
from starlette.middleware.gzip import GZipMiddleware
from devplacepy.cache import TTLCache
from devplacepy.config import (
PORT,
STATIC_DIR,
STATIC_VERSION,
UPLOADS_DIR,
ensure_data_dirs,
)
from devplacepy.db_client import (
db,
get_blocked_uids,
get_comment_counts_by_post_uids,
get_int_setting,
get_news_images_by_uids,
get_setting,
get_table,
get_user_post_count,
get_user_stars,
get_users_by_uids,
get_vote_counts,
interleave_by_author,
)
from devplacepy.responses import json_error, respond, wants_json
from devplacepy.routers import (
admin,
auth,
avatar,
awards,
bookmarks,
comments,
dbapi,
devrant,
docs,
feed,
follow,
game,
gists,
issues,
leaderboard,
media,
messages,
news,
notifications,
polls,
posts,
profile,
projects,
proxy,
push,
reactions,
relations,
seo,
uploads,
votes,
)
from devplacepy.schemas import LandingOut, ValidationErrorOut
from devplacepy.seo import base_seo_context, site_url, website_schema
from devplacepy.services import presence
from devplacepy.services.audit import record as audit
from devplacepy.services.correction import PENDING_SCOPE_KEY
from devplacepy.templating import templates
from devplacepy.utils import client_ip, get_current_user, safe_next, time_ago
from devplacepy_services.web.ingress import mount_ingress
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
)
logger = logging.getLogger(__name__)
_rate_limit_store = defaultdict(list)
RATE_LIMIT = int(os.environ.get("DEVPLACE_RATE_LIMIT", "60"))
RATE_WINDOW = 60
WEB_WORKERS = max(1, int(os.environ.get("DEVPLACE_WEB_WORKERS", "1")))
RATE_LIMIT_DISABLED = os.environ.get("DEVPLACE_DISABLE_RATE_LIMIT") == "1"
HOT_SETTINGS_TTL = 2.0
_hot_settings_value: dict = {}
_hot_settings_at = 0.0
_last_rate_sweep = 0.0
RATE_SWEEP_INTERVAL = 60.0
def _hot_settings() -> dict:
global _hot_settings_value, _hot_settings_at
now = time.monotonic()
if not _hot_settings_value or now - _hot_settings_at >= HOT_SETTINGS_TTL:
_hot_settings_value = {
"maintenance_mode": get_setting("maintenance_mode", "0"),
"rate_limit_per_minute": max(
1, get_int_setting("rate_limit_per_minute", RATE_LIMIT)
),
"rate_limit_window_seconds": max(
1, get_int_setting("rate_limit_window_seconds", RATE_WINDOW)
),
}
_hot_settings_at = now
return _hot_settings_value
def _sweep_rate_limit_store(window_start: float) -> None:
stale = [
ip
for ip, timestamps in _rate_limit_store.items()
if not timestamps or timestamps[-1] <= window_start
]
for ip in stale:
del _rate_limit_store[ip]
def _worker_rate_limit(limit: int) -> int:
return max(1, -(-limit // WEB_WORKERS))
INLINE_MEDIA_EXTENSIONS = {
".jpg",
".jpeg",
".png",
".gif",
".webp",
".bmp",
".tiff",
".mp4",
".webm",
".ogv",
".mov",
".m4v",
".mp3",
}
class UploadStaticFiles(StaticFiles):
async def get_response(self, path, scope):
response = await super().get_response(path, scope)
disposition = (
"inline"
if Path(path).suffix.lower() in INLINE_MEDIA_EXTENSIONS
else "attachment"
)
response.headers["Content-Disposition"] = disposition
response.headers["Cache-Control"] = "public, max-age=604800"
return response
class CachedStaticFiles(StaticFiles):
async def get_response(self, path, scope):
response = await super().get_response(path, scope)
if Path(path).name == "service-worker.js":
response.headers["Cache-Control"] = "no-cache"
else:
response.headers["Cache-Control"] = "public, max-age=31536000, immutable"
return response
class FallbackStaticFiles(StaticFiles):
async def get_response(self, path, scope):
response = await super().get_response(path, scope)
if Path(path).name == "service-worker.js":
response.headers["Cache-Control"] = "no-cache"
else:
response.headers["Cache-Control"] = "public, max-age=3600"
return response
@asynccontextmanager
async def lifespan(app: FastAPI):
ensure_data_dirs()
from devplacepy.services.statistics.tracking import start_visit_flusher
start_visit_flusher()
from devplacepy.config import SERVICE_LOCK_FILE
from devplacepy.services.dbapi.service import DbApiJobService
from devplacepy.services.manager import service_manager
from devplacepy.services.weblock import acquire_web_lock
web_lock_owner = acquire_web_lock(SERVICE_LOCK_FILE)
if web_lock_owner:
service_manager.register(DbApiJobService())
service_manager.set_lock_owner(True)
service_manager.supervise()
logger.info(f"DevPlace web started on port {PORT}")
yield
logger.info("Shutting down web...")
if web_lock_owner:
await service_manager.shutdown_all()
from devplacepy.services.statistics.tracking import flush_visits
flush_visits()
_AUTH_FORM_PAGES = {
"/auth/signup": ("signup.html", "Join DevPlace"),
"/auth/login": ("login.html", "Sign In"),
"/auth/forgot-password": ("forgot_password.html", "Reset Password"),
}
_FRIENDLY_ERRORS = {
("username", "too_short"): "Username must be between 3 and 32 characters",
("username", "too_long"): "Username must be between 3 and 32 characters",
("password", "too_short"): "Password must be at least 6 characters",
}
def _friendly_error(err):
field = err["loc"][-1] if err.get("loc") else ""
key = (field, err.get("type", "").replace("string_", ""))
if key in _FRIENDLY_ERRORS:
return _FRIENDLY_ERRORS[key]
msg = err.get("msg", "Invalid input")
prefix = "Value error, "
return msg[len(prefix) :] if msg.startswith(prefix) else msg
_home_cache = TTLCache(ttl=int(os.environ.get("DEVPLACE_HOME_CACHE_TTL", "60")), max_size=4)
def _landing_news():
cached = _home_cache.get("news")
if cached is not None:
return cached
articles = []
if "news" in db.tables:
news_table = get_table("news")
raw = list(
news_table.find(show_on_landing=1, order_by=["-synced_at"], _limit=6)
)
images_by_news = get_news_images_by_uids([a["uid"] for a in raw])
for a in raw:
articles.append(
{
"uid": a["uid"],
"slug": a.get("slug", ""),
"title": a.get("title", ""),
"description": (a.get("description", "") or "")[:250],
"url": a.get("url", ""),
"source_name": a.get("source_name", ""),
"grade": a.get("grade", 0),
"featured": a.get("featured", 0),
"synced_at": a.get("synced_at", "") or "",
"time_ago": time_ago(a["synced_at"]) if a.get("synced_at") else "",
"image_url": a.get("image_url", "") or images_by_news.get(a["uid"], ""),
}
)
_home_cache.set("news", articles)
return articles
def _landing_recent_posts(blocked):
if not blocked:
cached = _home_cache.get("posts")
if cached is not None:
return cached
posts = []
if "posts" in db.tables:
posts_table = get_table("posts")
fetch_limit = 24 if blocked else 6
raw_posts = list(
posts_table.find(deleted_at=None, order_by=["-created_at"], _limit=fetch_limit)
)
if blocked:
raw_posts = [p for p in raw_posts if p["user_uid"] not in blocked]
raw_posts = raw_posts[:6]
raw_posts = interleave_by_author(raw_posts)
if raw_posts:
post_uids = [p["uid"] for p in raw_posts]
author_uids = [p["user_uid"] for p in raw_posts]
authors = get_users_by_uids(author_uids)
comment_counts = get_comment_counts_by_post_uids(post_uids)
upvotes, downvotes = get_vote_counts(post_uids)
for p in raw_posts:
posts.append(
{
"post": p,
"author": authors.get(p["user_uid"]),
"time_ago": time_ago(p["created_at"]),
"comment_count": comment_counts.get(p["uid"], 0),
"stars": upvotes.get(p["uid"], 0) - downvotes.get(p["uid"], 0),
"slug": p.get("slug", "") or p["uid"],
}
)
if not blocked:
_home_cache.set("posts", posts)
return posts
def create_web_app() -> FastAPI:
from devplacepy_services.base.middleware import (
InternalCallCounterMiddleware,
RequestStatsMiddleware,
SanitizeErrorsMiddleware,
)
app = FastAPI(
title="DevPlace",
docs_url="/swagger",
redoc_url=None,
openapi_url="/openapi.json",
lifespan=lifespan,
)
app.add_middleware(SanitizeErrorsMiddleware)
app.add_middleware(InternalCallCounterMiddleware)
app.add_middleware(RequestStatsMiddleware)
app.mount(
"/static/uploads",
UploadStaticFiles(directory=str(UPLOADS_DIR), check_dir=False),
name="uploads",
)
app.mount(
f"/static/v{STATIC_VERSION}",
CachedStaticFiles(directory=str(STATIC_DIR)),
name="static_versioned",
)
app.mount("/static", FallbackStaticFiles(directory=str(STATIC_DIR)), name="static")
@app.exception_handler(404)
async def not_found(request: Request, exc):
if wants_json(request):
return json_error(404, "Not found")
seo_ctx = base_seo_context(
request,
title="Not Found - DevPlace",
description="The page you requested does not exist.",
robots="noindex",
)
return templates.TemplateResponse(
request,
"error.html",
{
**seo_ctx,
"request": request,
"error_code": 404,
"error_message": "Page not found",
},
status_code=404,
)
@app.exception_handler(500)
async def server_error(request: Request, exc):
logger.exception("500 error on %s %s", request.method, request.url.path)
if wants_json(request):
return json_error(500, "Internal server error")
seo_ctx = base_seo_context(
request,
title="Server Error - DevPlace",
description="Something went wrong.",
robots="noindex",
)
return templates.TemplateResponse(
request,
"error.html",
{
**seo_ctx,
"request": request,
"error_code": 500,
"error_message": "Internal server error",
},
status_code=500,
)
@app.exception_handler(RequestValidationError)
async def on_validation_error(request: Request, exc: RequestValidationError):
errors = [_friendly_error(e) for e in exc.errors()]
if wants_json(request):
fields: dict = {}
for raw, friendly in zip(exc.errors(), errors):
name = raw["loc"][-1] if raw.get("loc") else "_"
fields.setdefault(str(name), []).append(friendly)
return JSONResponse(
ValidationErrorOut(fields=fields, messages=errors).model_dump(mode="json"),
status_code=422,
)
path = request.url.path
page = _AUTH_FORM_PAGES.get(path)
if page is None and path.startswith("/auth/reset-password/"):
page = ("reset_password.html", "Set New Password")
if page:
template_name, title = page
context = {
**base_seo_context(request, title=title, robots="noindex,nofollow"),
"request": request,
"errors": errors,
}
try:
form = await request.form()
context.update({k: v for k, v in form.items() if isinstance(v, str)})
except Exception:
pass
if "token" in request.path_params:
context["token"] = request.path_params["token"]
return templates.TemplateResponse(
request, template_name, context, status_code=400
)
referer = safe_next(request.headers.get("referer"), "/feed")
return RedirectResponse(url=referer, status_code=303)
mount_ingress(app)
app.include_router(auth.router, prefix="/auth")
app.include_router(feed.router, prefix="/feed")
app.include_router(posts.router, prefix="/posts")
app.include_router(comments.router, prefix="/comments")
app.include_router(projects.router, prefix="/projects")
app.include_router(profile.router, prefix="/profile")
app.include_router(messages.router, prefix="/messages")
app.include_router(notifications.router, prefix="/notifications")
app.include_router(votes.router, prefix="/votes")
app.include_router(reactions.router, prefix="/reactions")
app.include_router(bookmarks.router, prefix="/bookmarks")
app.include_router(polls.router, prefix="/polls")
app.include_router(avatar.router, prefix="/avatar")
app.include_router(awards.router, prefix="/awards")
app.include_router(follow.router, prefix="/follow")
app.include_router(relations.router)
app.include_router(leaderboard.router, prefix="/leaderboard")
app.include_router(admin.router, prefix="/admin")
app.include_router(seo.router)
app.include_router(push.router)
app.include_router(docs.router)
app.include_router(issues.router, prefix="/issues")
app.include_router(gists.router, prefix="/gists")
app.include_router(news.router, prefix="/news")
app.include_router(uploads.router, prefix="/uploads")
app.include_router(media.router, prefix="/media")
app.include_router(proxy.router, prefix="/p")
app.include_router(devrant.router, prefix="/api")
app.include_router(dbapi.router, prefix="/dbapi")
app.include_router(game.router, prefix="/game")
@app.middleware("http")
async def await_pending_corrections(request: Request, call_next):
response = await call_next(request)
pending = request.scope.get(PENDING_SCOPE_KEY)
if pending:
await asyncio.gather(*pending, return_exceptions=True)
return response
@app.middleware("http")
async def add_security_headers(request: Request, call_next):
response = await call_next(request)
if not response.headers.get("X-Robots-Tag"):
response.headers["X-Robots-Tag"] = "index, follow"
response.headers["X-Content-Type-Options"] = "nosniff"
response.headers["Strict-Transport-Security"] = "max-age=31536000; includeSubDomains"
response.headers["Referrer-Policy"] = "strict-origin-when-cross-origin"
if not request.url.path.startswith("/p/"):
response.headers["X-Frame-Options"] = "DENY"
response.headers["Content-Security-Policy"] = (
"object-src 'none'; base-uri 'self'; "
"frame-ancestors 'none'; form-action 'self'"
)
if request.url.path.startswith("/admin"):
response.headers["Cache-Control"] = "no-store, no-cache, must-revalidate, max-age=0"
response.headers["Pragma"] = "no-cache"
response.headers["Expires"] = "0"
return response
@app.middleware("http")
async def rate_limit_middleware(request: Request, call_next):
if RATE_LIMIT_DISABLED:
return await call_next(request)
if request.method in (
"POST",
"PUT",
"DELETE",
"PATCH",
) and not request.url.path.startswith(("/openai", "/xmlrpc")):
settings = _hot_settings()
limit = _worker_rate_limit(settings["rate_limit_per_minute"])
window = settings["rate_limit_window_seconds"]
ip = client_ip(request, default="unknown")
now = time.time()
window_start = now - window
global _last_rate_sweep
if now - _last_rate_sweep >= RATE_SWEEP_INTERVAL:
_sweep_rate_limit_store(window_start)
_last_rate_sweep = now
timestamps = [t for t in _rate_limit_store.get(ip, ()) if t > window_start]
if len(timestamps) >= limit:
_rate_limit_store[ip] = timestamps
audit.record(
request,
"security.rate_limit.block",
result="denied",
summary=f"request from {ip} blocked by rate limit",
metadata={"ip": ip, "limit": limit, "window_seconds": window},
)
retry_after = {"Retry-After": str(window)}
if wants_json(request):
response = json_error(429, "Rate limit exceeded. Try again later.")
response.headers["Retry-After"] = str(window)
return response
return HTMLResponse(
"Rate limit exceeded. Try again later.",
status_code=429,
headers=retry_after,
)
timestamps.append(now)
_rate_limit_store[ip] = timestamps
return await call_next(request)
_MAINTENANCE_ALLOWED_PREFIXES = ("/static", "/avatar", "/auth", "/admin", "/openai")
@app.middleware("http")
async def maintenance_middleware(request: Request, call_next):
if _hot_settings()["maintenance_mode"] != "1":
return await call_next(request)
if request.url.path.startswith(_MAINTENANCE_ALLOWED_PREFIXES):
return await call_next(request)
user = get_current_user(request)
if user and user.get("role") == "Admin":
return await call_next(request)
message = get_setting(
"maintenance_message",
"DevPlace is undergoing scheduled maintenance. Please check back shortly.",
)
audit.record(
request,
"security.maintenance.block",
user=user,
result="denied",
summary="non-admin request blocked by maintenance mode",
)
if wants_json(request):
return json_error(503, message)
seo_ctx = base_seo_context(
request, title="Maintenance - DevPlace", description=message, robots="noindex"
)
return templates.TemplateResponse(
request,
"error.html",
{**seo_ctx, "request": request, "error_code": 503, "error_message": message},
status_code=503,
)
@app.middleware("http")
async def track_presence(request: Request, call_next):
path = request.url.path
if not path.startswith(("/static", "/avatar")):
user = get_current_user(request)
if user:
presence.touch(user["uid"])
return await call_next(request)
@app.middleware("http")
async def visit_statistics(request: Request, call_next):
from devplacepy.services.statistics.tracking import track_visit
response = await call_next(request)
track_visit(request, response.status_code)
return response
@app.middleware("http")
async def response_timing(request: Request, call_next):
start = time.perf_counter()
request.state.request_start = start
response = await call_next(request)
response.headers["X-Response-Time"] = f"{(time.perf_counter() - start) * 1000:.1f}ms"
return response
app.add_middleware(GZipMiddleware, minimum_size=512, compresslevel=6)
@app.get("/")
async def landing(request: Request):
user = get_current_user(request)
landing_articles = _landing_news()
blocked = get_blocked_uids(user["uid"]) if user else frozenset()
landing_posts = _landing_recent_posts(blocked)
base = site_url(request)
seo_ctx = base_seo_context(
request,
title="DevPlace - The Developer Social Network",
description="Track industry shifts. Discover bold releases. Share what you're building in an open, uncensored environment.",
breadcrumbs=[],
schemas=[website_schema(base)],
)
return respond(
request,
"landing.html",
{
**seo_ctx,
"request": request,
"user": user,
"is_authenticated": bool(user),
"user_post_count": get_user_post_count(user["uid"]) if user else 0,
"user_stars": get_user_stars(user["uid"]) if user else 0,
"landing_articles": landing_articles,
"landing_posts": landing_posts,
},
model=LandingOut,
)
return app
app = create_web_app()
Reference in New Issue View Git Blame Copy Permalink