use crate::core::parser::{UnifiedOperation, UnifiedSpec};
use anyhow::Result;
use std::fs;
use std::path::Path;
pub fn generate(
spec: &UnifiedSpec,
output_dir: &Path,
with_tests: bool,
with_validation: bool,
) -> Result<()> {
fs::create_dir_all(output_dir.join("app"))?;
fs::create_dir_all(output_dir.join("app/routers"))?;
fs::create_dir_all(output_dir.join("app/models"))?;
fs::create_dir_all(output_dir.join("app/core"))?;
if with_tests {
fs::create_dir_all(output_dir.join("tests"))?;
}
generate_main(spec, output_dir)?;
generate_routers(spec, output_dir)?;
generate_models(spec, output_dir)?;
generate_config(spec, output_dir)?;
generate_requirements(output_dir, with_validation)?;
if with_tests {
generate_tests(spec, output_dir)?;
}
Ok(())
}
fn generate_main(spec: &UnifiedSpec, output_dir: &Path) -> Result<()> {
let content = format!(
r#"""
FastAPI Application - {}
Generated by MicroRapid
"""
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from app.core.config import settings
{}
app = FastAPI(
title="{}",
description="{}",
version="{}",
docs_url="/docs",
redoc_url="/redoc",
)
# Configure CORS
app.add_middleware(
CORSMiddleware,
allow_origins=settings.CORS_ORIGINS,
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# Include routers
{}
@app.get("/")
async def root():
return {{"message": "API is running", "version": "{}"}}
@app.get("/health")
async def health_check():
return {{"status": "healthy"}}
"#,
spec.info.title,
generate_router_imports(&spec),
spec.info.title,
spec.info.description.as_deref().unwrap_or(""),
spec.info.version,
generate_router_includes(&spec),
spec.info.version
);
fs::write(output_dir.join("app/main.py"), content)?;
fs::write(output_dir.join("app/__init__.py"), "")?;
Ok(())
}
fn generate_routers(spec: &UnifiedSpec, output_dir: &Path) -> Result<()> {
let mut router_groups: std::collections::HashMap<String, Vec<&UnifiedOperation>> =
std::collections::HashMap::new();
for operation in &spec.operations {
let group = operation
.path
.split('/')
.nth(1)
.unwrap_or("default")
.to_string();
router_groups
.entry(group)
.or_insert_with(Vec::new)
.push(operation);
}
for (group_name, operations) in router_groups {
generate_router(&group_name, &operations, output_dir)?;
}
fs::write(output_dir.join("app/routers/__init__.py"), "")?;
Ok(())
}
fn generate_router(name: &str, operations: &[&UnifiedOperation], output_dir: &Path) -> Result<()> {
let mut route_handlers = String::new();
for op in operations {
route_handlers.push_str(&format!(
r#"
@router.{}("{}", {})
async def {}({}) -> {}:
"""
{}
TODO: Implement business logic
"""
# Example response
return {{"message": "Not implemented yet"}}
"#,
op.method.to_lowercase(),
op.path.replace(&format!("/{}", name), ""),
generate_operation_params(op),
to_snake_case(&op.operation_id),
generate_function_params(op),
generate_response_type(op),
op.summary.as_deref().unwrap_or(&op.operation_id)
));
}
let content = format!(
r#""""
{} Router
Generated by MicroRapid
"""
from typing import Optional, List, Dict, Any
from fastapi import APIRouter, Depends, HTTPException, Query, Path, Body
from pydantic import BaseModel
from app.models import schemas
from app.core import security
router = APIRouter(
prefix="/{}",
tags=["{}"],
)
{}
"#,
to_title_case(name),
name,
name,
route_handlers
);
fs::write(output_dir.join(format!("app/routers/{}.py", name)), content)?;
Ok(())
}
fn generate_models(_spec: &UnifiedSpec, output_dir: &Path) -> Result<()> {
let content = r#""""
Pydantic Models
Generated by MicroRapid
"""
from typing import Optional, List, Dict, Any
from pydantic import BaseModel, Field
from datetime import datetime
# Base models
class BaseResponse(BaseModel):
"""Base response model"""
success: bool = True
message: Optional[str] = None
class ErrorResponse(BaseModel):
"""Error response model"""
success: bool = False
error: str
detail: Optional[Dict[str, Any]] = None
# TODO: Add your models here based on OpenAPI schemas
# Example:
class UserBase(BaseModel):
"""User base model"""
username: str = Field(..., min_length=3, max_length=50)
email: str = Field(..., regex=r'^[\w\.-]+@[\w\.-]+\.\w+$')
full_name: Optional[str] = None
class UserCreate(UserBase):
"""User creation model"""
password: str = Field(..., min_length=8)
class User(UserBase):
"""User response model"""
id: int
created_at: datetime
updated_at: Optional[datetime] = None
class Config:
orm_mode = True
"#;
fs::write(output_dir.join("app/models/__init__.py"), "")?;
fs::write(output_dir.join("app/models/schemas.py"), content)?;
Ok(())
}
fn generate_config(_spec: &UnifiedSpec, output_dir: &Path) -> Result<()> {
let content = r#""""
Application Configuration
Generated by MicroRapid
"""
from typing import List, Optional
from pydantic import BaseSettings
class Settings(BaseSettings):
# Application
APP_NAME: str = "API"
VERSION: str = "1.0.0"
DEBUG: bool = False
# API
API_V1_PREFIX: str = "/api/v1"
# Security
SECRET_KEY: str = "your-secret-key-here"
ACCESS_TOKEN_EXPIRE_MINUTES: int = 30
# CORS
CORS_ORIGINS: List[str] = ["*"]
# Database (if needed)
DATABASE_URL: Optional[str] = None
class Config:
env_file = ".env"
case_sensitive = True
settings = Settings()
"#;
fs::write(output_dir.join("app/core/__init__.py"), "")?;
fs::write(output_dir.join("app/core/config.py"), content)?;
Ok(())
}
fn generate_requirements(output_dir: &Path, with_validation: bool) -> Result<()> {
let mut requirements = vec![
"fastapi>=0.104.0",
"uvicorn[standard]>=0.24.0",
"pydantic>=2.5.0",
"python-multipart>=0.0.6",
"httpx>=0.25.0",
];
if with_validation {
requirements.push("email-validator>=2.1.0");
requirements.push("python-jose[cryptography]>=3.3.0");
requirements.push("passlib[bcrypt]>=1.7.4");
}
let content = requirements.join("\n");
fs::write(output_dir.join("requirements.txt"), content)?;
Ok(())
}
fn generate_tests(_spec: &UnifiedSpec, output_dir: &Path) -> Result<()> {
let content = r#""""
Test Configuration
Generated by MicroRapid
"""
import pytest
from fastapi.testclient import TestClient
from app.main import app
@pytest.fixture
def client():
"""Create test client"""
return TestClient(app)
def test_root(client):
"""Test root endpoint"""
response = client.get("/")
assert response.status_code == 200
assert response.json()["message"] == "API is running"
def test_health(client):
"""Test health check"""
response = client.get("/health")
assert response.status_code == 200
assert response.json()["status"] == "healthy"
"#;
fs::write(output_dir.join("tests/__init__.py"), "")?;
fs::write(output_dir.join("tests/conftest.py"), content)?;
Ok(())
}
fn generate_router_imports(spec: &UnifiedSpec) -> String {
let mut imports = vec![];
let mut seen = std::collections::HashSet::new();
for op in &spec.operations {
let group = op.path.split('/').nth(1).unwrap_or("default").to_string();
if seen.insert(group.clone()) {
imports.push(format!("from app.routers import {}", group));
}
}
imports.join("\n")
}
fn generate_router_includes(spec: &UnifiedSpec) -> String {
let mut includes = vec![];
let mut seen = std::collections::HashSet::new();
for op in &spec.operations {
let group = op.path.split('/').nth(1).unwrap_or("default").to_string();
if seen.insert(group.clone()) {
includes.push(format!("app.include_router({}.router)", group));
}
}
includes.join("\n")
}
fn generate_operation_params(op: &UnifiedOperation) -> String {
let mut params = vec![];
params.push("response_model=schemas.BaseResponse".to_string());
if op.method == "POST" {
params.push("status_code=201".to_string());
}
if let Some(summary) = &op.summary {
params.push(format!("summary=\"{}\"", summary));
}
params.join(", ")
}
fn generate_function_params(op: &UnifiedOperation) -> String {
let mut params = vec![];
for param in &op.parameters {
if matches!(param.location, crate::core::parser::ParameterLocation::Path) {
params.push(format!(
"{}: str = Path(..., description=\"{}\")",
to_snake_case(¶m.name),
param.description.as_deref().unwrap_or("")
));
}
}
for param in &op.parameters {
if matches!(
param.location,
crate::core::parser::ParameterLocation::Query
) {
let param_type = if param.required {
"str"
} else {
"Optional[str] = None"
};
params.push(format!(
"{}: {} = Query({}, description=\"{}\")",
to_snake_case(¶m.name),
param_type,
if param.required { "..." } else { "None" },
param.description.as_deref().unwrap_or("")
));
}
}
if op.request_body.is_some() {
params.push("body: Dict[str, Any] = Body(...)".to_string());
}
if params.is_empty() {
String::new()
} else {
params.join(", ")
}
}
fn generate_response_type(_op: &UnifiedOperation) -> &'static str {
"schemas.BaseResponse"
}
fn to_snake_case(s: &str) -> String {
let mut result = String::new();
let mut prev_upper = false;
for (i, ch) in s.chars().enumerate() {
if ch.is_uppercase() && i > 0 && !prev_upper {
result.push('_');
}
result.push(ch.to_lowercase().next().unwrap());
prev_upper = ch.is_uppercase();
}
result
}
fn to_title_case(s: &str) -> String {
s.split('_')
.map(|word| {
let mut chars = word.chars();
match chars.next() {
None => String::new(),
Some(first) => first.to_uppercase().collect::<String>() + chars.as_str(),
}
})
.collect::<Vec<_>>()
.join(" ")
}