Implement notes management platform with FastAPI and SQLite

- Set up project structure with FastAPI
- Create database models for notes
- Implement Alembic migrations
- Create API endpoints for note CRUD operations
- Implement note export functionality (markdown, txt, pdf)
- Add health endpoint
- Set up linting with Ruff
This commit is contained in:
Automated Action 2025-06-04 08:13:43 +00:00
parent f261a39155
commit 355d2a84d5
26 changed files with 810 additions and 2 deletions

View File

@ -1,3 +1,96 @@
# FastAPI Application
# Notes Management Platform
This is a FastAPI application bootstrapped by BackendIM, the AI-powered backend generation platform.
A platform to store and download notes in various formats.
## Features
- Create, read, update, and delete notes
- Export notes in different formats:
- Markdown (.md)
- Plain text (.txt)
- PDF (.pdf)
- Search notes by title or content
## Tech Stack
- **Framework**: FastAPI
- **Database**: SQLite
- **ORM**: SQLAlchemy
- **Migrations**: Alembic
- **Export**: Markdown, WeasyPrint (PDF)
## Project Structure
```
.
├── alembic.ini # Alembic configuration
├── app/ # Application package
│ ├── api/ # API endpoints
│ ├── core/ # Core functionality (config, etc.)
│ ├── crud/ # CRUD operations
│ ├── db/ # Database setup
│ ├── models/ # SQLAlchemy models
│ ├── schemas/ # Pydantic schemas
│ ├── services/ # Business logic
│ └── utils/ # Utility functions
├── main.py # FastAPI application entry point
├── migrations/ # Alembic migrations
│ ├── env.py # Alembic environment
│ ├── script.py.mako # Migration script template
│ └── versions/ # Migration scripts
└── requirements.txt # Project dependencies
```
## Setup and Installation
### Prerequisites
- Python 3.8+
- pip (Python package installer)
### Installation
1. Clone the repository
2. Install dependencies:
```bash
pip install -r requirements.txt
```
3. Run database migrations:
```bash
alembic upgrade head
```
4. Start the application:
```bash
uvicorn main:app --host 0.0.0.0 --port 8000 --reload
```
## API Usage
Once the application is running, you can interact with the API using the Swagger UI at:
```
http://localhost:8000/docs
```
### Available Endpoints
- `GET /api/v1/notes` - List all notes (with optional search and pagination)
- `POST /api/v1/notes` - Create a new note
- `GET /api/v1/notes/{note_id}` - Get a specific note
- `PUT /api/v1/notes/{note_id}` - Update a note
- `DELETE /api/v1/notes/{note_id}` - Delete a note
- `POST /api/v1/notes/{note_id}/export` - Export a note to a specific format
- `GET /health` - Health check endpoint
## Development
### Linting
The project uses Ruff for linting and formatting:
```bash
ruff check .
ruff format .
```

117
alembic.ini Normal file
View File

@ -0,0 +1,117 @@
# A generic, single database configuration.
[alembic]
# path to migration scripts
script_location = migrations
# template used to generate migration file names; The default value is %%(rev)s_%%(slug)s
# Uncomment the line below if you want the files to be prepended with date and time
# see https://alembic.sqlalchemy.org/en/latest/tutorial.html#editing-the-ini-file
# for all available tokens
# file_template = %%(year)d_%%(month).2d_%%(day).2d_%%(hour).2d%%(minute).2d-%%(rev)s_%%(slug)s
# sys.path path, will be prepended to sys.path if present.
# defaults to the current working directory.
prepend_sys_path = .
# timezone to use when rendering the date within the migration file
# as well as the filename.
# If specified, requires the python-dateutil library that can be
# installed by adding `alembic[tz]` to the pip requirements
# string value is passed to dateutil.tz.gettz()
# leave blank for localtime
# timezone =
# max length of characters to apply to the
# "slug" field
# truncate_slug_length = 40
# set to 'true' to run the environment during
# the 'revision' command, regardless of autogenerate
# revision_environment = false
# set to 'true' to allow .pyc and .pyo files without
# a source .py file to be detected as revisions in the
# versions/ directory
# sourceless = false
# version location specification; This defaults
# to migrations/versions. When using multiple version
# directories, initial revisions must be specified with --version-path.
# The path separator used here should be the separator specified by "version_path_separator" below.
# version_locations = %(here)s/bar:%(here)s/bat:migrations/versions
# version path separator; As mentioned above, this is the character used to split
# version_locations. The default within new alembic.ini files is "os", which uses os.pathsep.
# If this key is omitted entirely, it falls back to the legacy behavior of splitting on spaces and/or commas.
# Valid values for version_path_separator are:
#
# version_path_separator = :
# version_path_separator = ;
# version_path_separator = space
version_path_separator = os # Use os.pathsep. Default configuration used for new projects.
# set to 'true' to search source files recursively
# in each "version_locations" directory
# new in Alembic version 1.10
# recursive_version_locations = false
# the output encoding used when revision files
# are written from script.py.mako
# output_encoding = utf-8
# SQLite URL example
sqlalchemy.url = sqlite:////app/storage/db/db.sqlite
[post_write_hooks]
# post_write_hooks defines scripts or Python functions that are run
# on newly generated revision scripts. See the documentation for further
# detail and examples
# format using "black" - use the console_scripts runner, against the "black" entrypoint
# hooks = black
# black.type = console_scripts
# black.entrypoint = black
# black.options = -l 79 REVISION_SCRIPT_FILENAME
# lint with attempts to fix using "ruff" - use the exec runner, execute a binary
# hooks = ruff
# ruff.type = exec
# ruff.executable = %(here)s/.venv/bin/ruff
# ruff.options = --fix REVISION_SCRIPT_FILENAME
# Logging configuration
[loggers]
keys = root,sqlalchemy,alembic
[handlers]
keys = console
[formatters]
keys = generic
[logger_root]
level = WARN
handlers = console
qualname =
[logger_sqlalchemy]
level = WARN
handlers =
qualname = sqlalchemy.engine
[logger_alembic]
level = INFO
handlers =
qualname = alembic
[handler_console]
class = StreamHandler
args = (sys.stderr,)
level = NOTSET
formatter = generic
[formatter_generic]
format = %(levelname)-5.5s [%(name)s] %(message)s
datefmt = %H:%M:%S

1
app/__init__.py Normal file
View File

@ -0,0 +1 @@
# Notes Management Platform Application

1
app/api/__init__.py Normal file
View File

@ -0,0 +1 @@
# API package

View File

@ -0,0 +1 @@
# API endpoints package

100
app/api/endpoints/notes.py Normal file
View File

@ -0,0 +1,100 @@
from typing import List, Optional
from fastapi import APIRouter, Depends, HTTPException, Query, status
from sqlalchemy.orm import Session
from app.crud import note as note_crud
from app.db.session import get_db
from app.schemas.note import Note, NoteCreate, NoteExport, NoteUpdate
from app.services.export import export_note
router = APIRouter()
@router.get("/", response_model=List[Note])
async def read_notes(
skip: int = 0,
limit: int = 100,
search: Optional[str] = Query(None, description="Search query for title and content"),
db: Session = Depends(get_db)
):
"""
Retrieve all notes, with optional pagination and search.
"""
if search:
notes = note_crud.search_notes(db, search, skip=skip, limit=limit)
else:
notes = note_crud.get_notes(db, skip=skip, limit=limit)
return notes
@router.post("/", response_model=Note, status_code=status.HTTP_201_CREATED)
async def create_note(
note: NoteCreate,
db: Session = Depends(get_db)
):
"""
Create a new note.
"""
return note_crud.create_note(db=db, note=note)
@router.get("/{note_id}", response_model=Note)
async def read_note(
note_id: int,
db: Session = Depends(get_db)
):
"""
Get a specific note by ID.
"""
db_note = note_crud.get_note(db, note_id=note_id)
if db_note is None:
raise HTTPException(status_code=404, detail="Note not found")
return db_note
@router.put("/{note_id}", response_model=Note)
async def update_note(
note_id: int,
note: NoteUpdate,
db: Session = Depends(get_db)
):
"""
Update a note.
"""
db_note = note_crud.update_note(db, note_id=note_id, note=note)
if db_note is None:
raise HTTPException(status_code=404, detail="Note not found")
return db_note
@router.delete("/{note_id}", status_code=status.HTTP_204_NO_CONTENT, response_model=None)
async def delete_note(
note_id: int,
db: Session = Depends(get_db)
):
"""
Delete a note.
"""
success = note_crud.delete_note(db, note_id=note_id)
if not success:
raise HTTPException(status_code=404, detail="Note not found")
return None
@router.post("/{note_id}/export", status_code=status.HTTP_200_OK)
async def export_note_endpoint(
note_id: int,
export_options: NoteExport,
db: Session = Depends(get_db)
):
"""
Export a note to the specified format.
"""
db_note = note_crud.get_note(db, note_id=note_id)
if db_note is None:
raise HTTPException(status_code=404, detail="Note not found")
file_path = export_note(db_note, export_options.format)
return {"message": "Note exported successfully", "file_path": str(file_path)}

22
app/api/routes.py Normal file
View File

@ -0,0 +1,22 @@
from fastapi import APIRouter, status
from fastapi.responses import JSONResponse
from app.api.endpoints import notes
from app.core.config import settings
api_router = APIRouter()
# Include all endpoint routers
api_router.include_router(notes.router, prefix=f"{settings.API_V1_STR}/notes", tags=["notes"])
# Health endpoint
@api_router.get("/health", tags=["health"])
async def health_check():
"""
Health check endpoint to verify that the API is running.
"""
return JSONResponse(
status_code=status.HTTP_200_OK,
content={"status": "healthy"}
)

1
app/core/__init__.py Normal file
View File

@ -0,0 +1 @@
# Core functionality package

29
app/core/config.py Normal file
View File

@ -0,0 +1,29 @@
from pathlib import Path
from pydantic_settings import BaseSettings
class Settings(BaseSettings):
PROJECT_NAME: str = "Notes Management Platform"
PROJECT_DESCRIPTION: str = "A platform to store and download notes in various formats"
VERSION: str = "0.1.0"
API_V1_STR: str = "/api/v1"
# Database settings
DB_DIR: Path = Path("/app/storage/db")
SQLALCHEMY_DATABASE_URL: str = f"sqlite:///{DB_DIR}/db.sqlite"
# Storage path for generated files
STORAGE_DIR: Path = Path("/app/storage")
EXPORT_DIR: Path = STORAGE_DIR / "exports"
class Config:
env_file = ".env"
case_sensitive = True
settings = Settings()
# Ensure required directories exist
settings.DB_DIR.mkdir(parents=True, exist_ok=True)
settings.EXPORT_DIR.mkdir(parents=True, exist_ok=True)

1
app/crud/__init__.py Normal file
View File

@ -0,0 +1 @@
# CRUD operations package

52
app/crud/note.py Normal file
View File

@ -0,0 +1,52 @@
from typing import List, Optional
from sqlalchemy.orm import Session
from app.models.note import Note
from app.schemas.note import NoteCreate, NoteUpdate
def get_note(db: Session, note_id: int) -> Optional[Note]:
return db.query(Note).filter(Note.id == note_id).first()
def get_notes(db: Session, skip: int = 0, limit: int = 100) -> List[Note]:
return db.query(Note).order_by(Note.updated_at.desc()).offset(skip).limit(limit).all()
def search_notes(db: Session, query: str, skip: int = 0, limit: int = 100) -> List[Note]:
return db.query(Note).filter(
(Note.title.contains(query)) | (Note.content.contains(query))
).order_by(Note.updated_at.desc()).offset(skip).limit(limit).all()
def create_note(db: Session, note: NoteCreate) -> Note:
db_note = Note(title=note.title, content=note.content)
db.add(db_note)
db.commit()
db.refresh(db_note)
return db_note
def update_note(db: Session, note_id: int, note: NoteUpdate) -> Optional[Note]:
db_note = db.query(Note).filter(Note.id == note_id).first()
if not db_note:
return None
update_data = note.model_dump(exclude_unset=True)
for key, value in update_data.items():
setattr(db_note, key, value)
db.commit()
db.refresh(db_note)
return db_note
def delete_note(db: Session, note_id: int) -> bool:
db_note = db.query(Note).filter(Note.id == note_id).first()
if not db_note:
return False
db.delete(db_note)
db.commit()
return True

1
app/db/__init__.py Normal file
View File

@ -0,0 +1 @@
# Database package

26
app/db/session.py Normal file
View File

@ -0,0 +1,26 @@
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from app.core.config import settings
# Create engine
engine = create_engine(
settings.SQLALCHEMY_DATABASE_URL,
connect_args={"check_same_thread": False} # Only needed for SQLite
)
# Create SessionLocal class
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
# Create Base class
Base = declarative_base()
# Dependency to get DB session
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()

2
app/models/__init__.py Normal file
View File

@ -0,0 +1,2 @@
# Import all models here to ensure they are registered with SQLAlchemy
from app.models.note import Note # noqa: F401

13
app/models/note.py Normal file
View File

@ -0,0 +1,13 @@
from sqlalchemy import Column, DateTime, Integer, String, Text, func
from app.db.session import Base
class Note(Base):
__tablename__ = "notes"
id = Column(Integer, primary_key=True, index=True)
title = Column(String(255), index=True)
content = Column(Text, nullable=False)
created_at = Column(DateTime(timezone=True), server_default=func.now())
updated_at = Column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now())

1
app/schemas/__init__.py Normal file
View File

@ -0,0 +1 @@
# Schemas package

35
app/schemas/note.py Normal file
View File

@ -0,0 +1,35 @@
from datetime import datetime
from typing import Literal, Optional
from pydantic import BaseModel, Field
class NoteBase(BaseModel):
title: str = Field(..., min_length=1, max_length=255)
content: str = Field(..., min_length=1)
class NoteCreate(NoteBase):
pass
class NoteUpdate(BaseModel):
title: Optional[str] = Field(None, min_length=1, max_length=255)
content: Optional[str] = Field(None, min_length=1)
class NoteInDB(NoteBase):
id: int
created_at: datetime
updated_at: datetime
class Config:
from_attributes = True
class Note(NoteInDB):
pass
class NoteExport(BaseModel):
format: Literal["markdown", "txt", "pdf"] = "markdown"

1
app/services/__init__.py Normal file
View File

@ -0,0 +1 @@
# Services package

85
app/services/export.py Normal file
View File

@ -0,0 +1,85 @@
import os
import tempfile
import uuid
from datetime import datetime
from pathlib import Path
from typing import Literal
import markdown
from weasyprint import HTML
from app.core.config import settings
from app.models.note import Note
def export_note(note: Note, format_type: Literal["markdown", "txt", "pdf"]) -> Path:
"""
Export a note to the specified format.
Args:
note: The note to export
format_type: The format to export to (markdown, txt, pdf)
Returns:
Path to the exported file
"""
# Create a unique filename
timestamp = datetime.now().strftime("%Y%m%d%H%M%S")
unique_id = str(uuid.uuid4())[:8]
if format_type == "markdown":
return _export_markdown(note, timestamp, unique_id)
elif format_type == "txt":
return _export_txt(note, timestamp, unique_id)
elif format_type == "pdf":
return _export_pdf(note, timestamp, unique_id)
else:
raise ValueError(f"Unsupported format: {format_type}")
def _export_markdown(note: Note, timestamp: str, unique_id: str) -> Path:
"""Export note to Markdown format."""
filename = f"{note.title.replace(' ', '_')}_{timestamp}_{unique_id}.md"
export_path = settings.EXPORT_DIR / filename
with open(export_path, "w") as f:
f.write(f"# {note.title}\n\n")
f.write(note.content)
return export_path
def _export_txt(note: Note, timestamp: str, unique_id: str) -> Path:
"""Export note to plain text format."""
filename = f"{note.title.replace(' ', '_')}_{timestamp}_{unique_id}.txt"
export_path = settings.EXPORT_DIR / filename
with open(export_path, "w") as f:
f.write(f"Title: {note.title}\n\n")
f.write(note.content)
return export_path
def _export_pdf(note: Note, timestamp: str, unique_id: str) -> Path:
"""Export note to PDF format."""
filename = f"{note.title.replace(' ', '_')}_{timestamp}_{unique_id}.pdf"
export_path = settings.EXPORT_DIR / filename
# Convert markdown to HTML
html_content = f"<h1>{note.title}</h1>\n{markdown.markdown(note.content)}"
# Create a temporary HTML file
with tempfile.NamedTemporaryFile(suffix='.html', delete=False) as temp_html:
temp_html.write(html_content.encode('utf-8'))
temp_html_path = temp_html.name
try:
# Convert HTML to PDF
HTML(string=html_content).write_pdf(export_path)
finally:
# Clean up temporary file
if os.path.exists(temp_html_path):
os.unlink(temp_html_path)
return export_path

1
app/utils/__init__.py Normal file
View File

@ -0,0 +1 @@
# Utilities package

49
main.py Normal file
View File

@ -0,0 +1,49 @@
import uvicorn
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from fastapi.openapi.utils import get_openapi
from app.api.routes import api_router
from app.core.config import settings
app = FastAPI(
title=settings.PROJECT_NAME,
description=settings.PROJECT_DESCRIPTION,
version=settings.VERSION,
docs_url="/docs",
redoc_url="/redoc",
)
# Add CORS middleware
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# Include all API routes
app.include_router(api_router)
# Custom OpenAPI schema to ensure it's at the root path
def custom_openapi():
if app.openapi_schema:
return app.openapi_schema
openapi_schema = get_openapi(
title=app.title,
version=app.version,
description=app.description,
routes=app.routes,
)
app.openapi_schema = openapi_schema
return app.openapi_schema
app.openapi = custom_openapi
if __name__ == "__main__":
uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True)

80
migrations/env.py Normal file
View File

@ -0,0 +1,80 @@
from logging.config import fileConfig
from alembic import context
from sqlalchemy import engine_from_config, pool
from app.db.session import Base
# this is the Alembic Config object, which provides
# access to the values within the .ini file in use.
config = context.config
# Interpret the config file for Python logging.
# This line sets up loggers basically.
if config.config_file_name is not None:
fileConfig(config.config_file_name)
# add your model's MetaData object here
# for 'autogenerate' support
target_metadata = Base.metadata
# other values from the config, defined by the needs of env.py,
# can be acquired:
# my_important_option = config.get_main_option("my_important_option")
# ... etc.
def run_migrations_offline() -> None:
"""Run migrations in 'offline' mode.
This configures the context with just a URL
and not an Engine, though an Engine is acceptable
here as well. By skipping the Engine creation
we don't even need a DBAPI to be available.
Calls to context.execute() here emit the given string to the
script output.
"""
url = config.get_main_option("sqlalchemy.url")
context.configure(
url=url,
target_metadata=target_metadata,
literal_binds=True,
dialect_opts={"paramstyle": "named"},
)
with context.begin_transaction():
context.run_migrations()
def run_migrations_online() -> None:
"""Run migrations in 'online' mode.
In this scenario we need to create an Engine
and associate a connection with the context.
"""
connectable = engine_from_config(
config.get_section(config.config_ini_section, {}),
prefix="sqlalchemy.",
poolclass=pool.NullPool,
)
with connectable.connect() as connection:
is_sqlite = connection.dialect.name == "sqlite"
context.configure(
connection=connection,
target_metadata=target_metadata,
render_as_batch=is_sqlite, # Key configuration for SQLite
)
with context.begin_transaction():
context.run_migrations()
if context.is_offline_mode():
run_migrations_offline()
else:
run_migrations_online()

24
migrations/script.py.mako Normal file
View File

@ -0,0 +1,24 @@
"""${message}
Revision ID: ${up_revision}
Revises: ${down_revision | comma,n}
Create Date: ${create_date}
"""
from alembic import op
import sqlalchemy as sa
${imports if imports else ""}
# revision identifiers, used by Alembic.
revision = ${repr(up_revision)}
down_revision = ${repr(down_revision)}
branch_labels = ${repr(branch_labels)}
depends_on = ${repr(depends_on)}
def upgrade() -> None:
${upgrades if upgrades else "pass"}
def downgrade() -> None:
${downgrades if downgrades else "pass"}

View File

@ -0,0 +1,43 @@
"""initial migration - create notes table
Revision ID: 001
Revises:
Create Date: 2023-11-01
"""
import sqlalchemy as sa
from alembic import op
# revision identifiers, used by Alembic.
revision = '001'
down_revision = None
branch_labels = None
depends_on = None
def upgrade() -> None:
# Create notes table
op.create_table(
'notes',
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('title', sa.String(length=255), nullable=False),
sa.Column('content', sa.Text(), nullable=False),
sa.Column('created_at', sa.DateTime(timezone=True),
server_default=sa.text('(CURRENT_TIMESTAMP)'), nullable=False),
sa.Column('updated_at', sa.DateTime(timezone=True),
server_default=sa.text('(CURRENT_TIMESTAMP)'), nullable=False),
sa.PrimaryKeyConstraint('id')
)
# Create index on title for faster lookups
op.create_index(op.f('ix_notes_id'), 'notes', ['id'], unique=False)
op.create_index(op.f('ix_notes_title'), 'notes', ['title'], unique=False)
def downgrade() -> None:
# Drop indexes
op.drop_index(op.f('ix_notes_title'), table_name='notes')
op.drop_index(op.f('ix_notes_id'), table_name='notes')
# Drop table
op.drop_table('notes')

18
pyproject.toml Normal file
View File

@ -0,0 +1,18 @@
[tool.ruff]
line-length = 100
target-version = "py38"
[tool.ruff.lint]
select = [
"E", # pycodestyle errors
"F", # pyflakes
"I", # isort
"W", # pycodestyle warnings
]
ignore = []
[tool.ruff.format]
quote-style = "double"
indent-style = "space"
skip-magic-trailing-comma = false
line-ending = "auto"

11
requirements.txt Normal file
View File

@ -0,0 +1,11 @@
fastapi>=0.104.1
uvicorn>=0.24.0
sqlalchemy>=2.0.23
alembic>=1.12.1
pydantic>=2.4.2
pydantic-settings>=2.0.3
python-multipart>=0.0.6
jinja2>=3.1.2
markdown>=3.5.1
weasyprint>=60.1
ruff>=0.1.5