122 lines
3.6 KiB
Python
122 lines
3.6 KiB
Python
from fastapi import FastAPI
|
|
from pydantic import BaseModel, EmailStr
|
|
from typing import List, Optional
|
|
from contextlib import asynccontextmanager
|
|
import sqlalchemy
|
|
from sqlalchemy import Column, String, Boolean, Text
|
|
from sqlalchemy.ext.declarative import declarative_base
|
|
import databases
|
|
|
|
DATABASE_URL = "mysql+pymysql://username:password@localhost:3306/mydatabase"
|
|
|
|
database = databases.Database(DATABASE_URL)
|
|
metadata = sqlalchemy.MetaData()
|
|
Base = declarative_base()
|
|
|
|
# -------------------- SQLAlchemy Models --------------------
|
|
class ProjectModel(Base):
|
|
__tablename__ = "projects"
|
|
id = Column(String(36), primary_key=True)
|
|
name = Column(String(255), nullable=False)
|
|
role = Column(String(255), nullable=False)
|
|
description = Column(Text, nullable=False)
|
|
highlights = Column(Text) # store as JSON string
|
|
stack = Column(Text) # store as JSON string
|
|
website = Column(String(255), nullable=True)
|
|
github = Column(String(255), nullable=True)
|
|
featured = Column(Boolean, default=False)
|
|
|
|
class StackCategoryModel(Base):
|
|
__tablename__ = "stack_categories"
|
|
id = Column(String(36), primary_key=True)
|
|
title = Column(String(255), nullable=False)
|
|
items = Column(Text) # JSON string
|
|
|
|
class UserModel(Base):
|
|
__tablename__ = "users"
|
|
id = Column(String(36), primary_key=True)
|
|
email = Column(String(255), nullable=False, unique=True)
|
|
name = Column(String(255), nullable=False)
|
|
|
|
# -------------------- Pydantic Schemas --------------------
|
|
|
|
class Project(BaseModel):
|
|
id: Optional[str]
|
|
name: str
|
|
role: str
|
|
description: str
|
|
highlights: List[str]
|
|
stack: List[str]
|
|
website: Optional[str] = None
|
|
github: Optional[str] = None
|
|
featured: bool
|
|
|
|
class StackCategory(BaseModel):
|
|
id: Optional[str]
|
|
title: str
|
|
items: List[str]
|
|
|
|
class User(BaseModel):
|
|
id: Optional[str]
|
|
email: EmailStr
|
|
name: str
|
|
|
|
|
|
# -------------------- FastAPI App --------------------
|
|
|
|
@asynccontextmanager
|
|
async def lifespan(app: FastAPI):
|
|
await database.connect()
|
|
yield
|
|
await database.disconnect()
|
|
|
|
app = FastAPI(title="atticl.com API", version="1.0", docs_url=None, redoc_url=None, openapi_url=None, lifespan=lifespan)
|
|
|
|
# -------------------- Helper Functions --------------------
|
|
|
|
import json
|
|
import uuid
|
|
|
|
def to_json_str(lst: List[str]) -> str:
|
|
return json.dumps(lst)
|
|
|
|
def from_json_str(s: str) -> List[str]:
|
|
return json.loads(s)
|
|
|
|
# -------------------- Project Endpoints --------------------
|
|
|
|
@app.post("/projects/", response_model=Project)
|
|
async def create_project(project: Project):
|
|
project.id = str(uuid.uuid4())
|
|
query = ProjectModel.__table__.insert().values(
|
|
id=project.id,
|
|
name=project.name,
|
|
role=project.role,
|
|
description=project.description,
|
|
highlights=to_json_str(project.highlights),
|
|
stack=to_json_str(project.stack),
|
|
website=project.website,
|
|
github=project.github,
|
|
featured=project.featured
|
|
)
|
|
await database.execute(query)
|
|
return project
|
|
|
|
@app.get("/projects/", response_model=List[Project])
|
|
async def list_projects():
|
|
query = ProjectModel.__table__.select()
|
|
rows = await database.fetch_all(query)
|
|
result = []
|
|
for row in rows:
|
|
result.append(Project(
|
|
id=row["id"],
|
|
name=row["name"],
|
|
role=row["role"],
|
|
description=row["description"],
|
|
highlights=from_json_str(row["highlights"]),
|
|
stack=from_json_str(row["stack"]),
|
|
website=row["website"],
|
|
github=row["github"],
|
|
featured=row["featured"]
|
|
))
|
|
return result |