README: Local Run & Deployment
# Cgroup Monorepo (frontend + backend)
Prereqs: Node 18+, Python 3.11+, Docker, Docker Compose, PostgreSQL (optional if using Docker)
Environment:
- DATABASE_URL=postgresql+psycopg://user:pass@db:5432/cgroup
- JWT_SECRET=CHANGE_ME
- SMTP_URL=smtp+tls://user:pass@smtp.mailhost.com:587
- CORS_ORIGIN=https://cgroup.example.com
Frontend (Next.js + Tailwind + shadcn/ui):
- pnpm install
- pnpm dev
- i18n (next-intl), theme (next-themes), SEO meta in _app and Head.
- Pages: Home, About, Services, Case Studies, Blog (SSG), Contact
- Auth pages: Login, Register, Forgot, Reset (client calls backend).
Backend (FastAPI):
- uvicorn app.main:app --reload
- Alembic migrations: alembic upgrade head
- Rate limiting (slowapi), JWT (jose), bcrypt, CORS restricted.
Docker:
- docker compose up --build
- Services: web (Next.js), api (FastAPI), db (Postgres), worker (optional), mail (optional)
Production:
- Configure HTTPS (reverse proxy)
- Set secure cookies (HttpOnly, SameSite=Lax/Strict), HSTS
- Rotate JWT secret, enforce strong password, 2FA optional
FastAPI: main.py (secure API, JWT, RBAC, rate limiting)
from fastapi import FastAPI, Depends, HTTPException, Request, BackgroundTasks
from fastapi.security import OAuth2PasswordRequestForm
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel, EmailStr, constr
from datetime import datetime, timedelta
from jose import jwt, JWTError
from passlib.hash import bcrypt
from slowapi import Limiter
from slowapi.util import get_remote_address
from sqlalchemy.orm import Session
from app.db import get_db, User, Report, BlogPost, Activity
import smtplib, ssl, os
JWT_SECRET = os.environ["JWT_SECRET"]
ACCESS_MIN = 15
REFRESH_DAYS = 30
limiter = Limiter(key_func=get_remote_address)
app = FastAPI(title="Cgroup API", version="1.0")
app.add_middleware(
CORSMiddleware,
allow_origins=[os.environ.get("CORS_ORIGIN", "")],
allow_methods=["POST","GET","PUT","DELETE","OPTIONS"],
allow_headers=["*"],
allow_credentials=True
)
class RegisterIn(BaseModel):
name: constr(min_length=1, max_length=100)
email: EmailStr
password: constr(min_length=8)
class TokenOut(BaseModel):
access_token: str
token_type: str = "bearer"
def make_token(data: dict, minutes: int):
to_encode = data | {"exp": datetime.utcnow() + timedelta(minutes=minutes)}
return jwt.encode(to_encode, JWT_SECRET, algorithm="HS256")
def get_current_user(db: Session = Depends(get_db), token: str | None = None, req: Request = None):
token = req.cookies.get("access_token")
if not token: raise HTTPException(401)
try:
payload = jwt.decode(token, JWT_SECRET, algorithms=["HS256"])
user = db.query(User).get(payload["sub"])
if not user: raise HTTPException(401)
return user
except JWTError:
raise HTTPException(401)
def require_role(*roles):
def wrapper(user=Depends(get_current_user)):
if user.role not in roles:
raise HTTPException(403, "Forbidden")
return user
return wrapper
@app.post("/auth/register", response_model=TokenOut)
@limiter.limit("5/minute")
def register(data: RegisterIn, db: Session = Depends(get_db)):
if db.query(User).filter_by(email=data.email).first():
raise HTTPException(400, "Email exists")
user = User(name=data.name, email=data.email, pass_hash=bcrypt.hash(data.password), role="Client")
db.add(user); db.commit(); db.refresh(user)
access = make_token({"sub": user.id, "role": user.role}, ACCESS_MIN)
refresh = make_token({"sub": user.id, "type": "refresh"}, 60*24*REFRESH_DAYS)
from fastapi.responses import JSONResponse
res = JSONResponse({"access_token": access})
# HttpOnly cookies
res.set_cookie("access_token", access, httponly=True, samesite="lax", secure=True)
res.set_cookie("refresh_token", refresh, httponly=True, samesite="lax", secure=True)
return res
@app.post("/auth/login", response_model=TokenOut)
@limiter.limit("10/minute")
def login(form: OAuth2PasswordRequestForm = Depends(), db: Session = Depends(get_db)):
user = db.query(User).filter_by(email=form.username).first()
if not user or not bcrypt.verify(form.password, user.pass_hash):
# record failed login
db.add(Activity(actor="system", action="failed_login", details=user.email if user else "unknown"))
db.commit()
raise HTTPException(401, "Invalid credentials")
access = make_token({"sub": user.id, "role": user.role}, ACCESS_MIN)
refresh = make_token({"sub": user.id, "type": "refresh"}, 60*24*REFRESH_DAYS)
from fastapi.responses import JSONResponse
res = JSONResponse({"access_token": access})
res.set_cookie("access_token", access, httponly=True, samesite="lax", secure=True)
res.set_cookie("refresh_token", refresh, httponly=True, samesite="lax", secure=True)
return res
@app.post("/auth/refresh")
def refresh(req: Request, db: Session = Depends(get_db)):
from fastapi.responses import JSONResponse
token = req.cookies.get("refresh_token")
if not token: raise HTTPException(401)
try:
payload = jwt.decode(token, JWT_SECRET, algorithms=["HS256"])
user = db.query(User).get(payload["sub"])
if not user: raise HTTPException(401)
access = make_token({"sub": user.id, "role": user.role}, ACCESS_MIN)
res = JSONResponse({"access_token": access})
res.set_cookie("access_token", access, httponly=True, samesite="lax", secure=True)
return res
except JWTError:
raise HTTPException(401)
class ContactIn(BaseModel):
name: str
company: str | None = None
email: EmailStr
message: constr(min_length=1, max_length=2000)
@app.post("/contact")
@limiter.limit("3/minute")
def contact(data: ContactIn, background: BackgroundTasks):
smtp_url = os.environ["SMTP_URL"]
def send_mail():
import urllib.parse
parsed = urllib.parse.urlparse(smtp_url)
user, pw = parsed.username, parsed.password
host, port = parsed.hostname, parsed.port
ctx = ssl.create_default_context()
with smtplib.SMTP(host, port) as s:
s.starttls(context=ctx)
if user: s.login(user, pw)
fromaddr = data.email
toaddr = "dobryjruslan71@gmail.com"
body = f"From: {data.name} <{data.email}>\nCompany: {data.company}\n\n{data.message}"
s.sendmail(fromaddr, [toaddr], f"Subject: Cgroup Contact\n\n{body}")
background.add_task(send_mail)
return {"ok": True}
# RBAC-protected endpoints for reports, users, blog, etc. ...
SQLAlchemy Models + Alembic Migration
# app/db.py
from sqlalchemy import Column, Integer, String, DateTime, Text, ForeignKey
from sqlalchemy.orm import declarative_base, relationship, sessionmaker
from sqlalchemy import create_engine
import os, datetime
DATABASE_URL = os.environ["DATABASE_URL"]
engine = create_engine(DATABASE_URL, pool_pre_ping=True, future=True)
SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False, future=True)
Base = declarative_base()
def get_db():
db = SessionLocal()
try: yield db
finally: db.close()
class User(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True)
name = Column(String(120))
email = Column(String(160), unique=True, index=True)
pass_hash = Column(String(255))
role = Column(String(20), default="Client")
class Report(Base):
__tablename__ = "reports"
id = Column(Integer, primary_key=True)
title = Column(String(200))
date = Column(DateTime, default=datetime.datetime.utcnow)
status = Column(String(30))
url = Column(Text) # Signed URL
user_id = Column(Integer, ForeignKey("users.id"))
user = relationship("User")
class BlogPost(Base):
__tablename__ = "blog_posts"
id = Column(Integer, primary_key=True)
title = Column(String(200))
body_md = Column(Text)
lang = Column(String(2), default="en")
author_id = Column(Integer, ForeignKey("users.id"))
created_at = Column(DateTime, default=datetime.datetime.utcnow)
class Activity(Base):
__tablename__ = "activity"
id = Column(Integer, primary_key=True)
actor = Column(String(160))
action = Column(String(100))
details = Column(Text)
at = Column(DateTime, default=datetime.datetime.utcnow)
# Alembic: generate migration and upgrade.