Add demo fastapi routes and models

This commit is contained in:
linuskmr 2021-12-21 21:00:58 +01:00
parent 3987c69a25
commit e45bf760b6
13 changed files with 372 additions and 10 deletions

32
backend/README.md Normal file
View file

@ -0,0 +1,32 @@
# Juggl API
## Installation
Minimum Python version: 3.10
Create virtual environment:
```
$ python3.10 -m venv/ ./venv/
```
Activate virtual environment (Linux):
```
$ source venv/bin/activate
```
Activate virtual environment (Windows):
```
$ .venv\Scripts\activate
```
## Guidelines
All routes and should be async for maximum performance
Always specify the `response_model` in the route decorator and use the same type as type hint for the function return
type. FastAPI is unfortunately not yet smart enough to determine the `response_model` from the return type hint.
Maybe we should consider making a pull request...

View file

@ -1,8 +0,0 @@
from fastapi import FastAPI
app = FastAPI()
@app.get("/")
async def root():
return {"message": "Hello World"}

View file

@ -1,2 +1,3 @@
fastapi uvicorn~=0.15.0
uvicorn fastapi~=0.70.0
pydantic~=1.8.2

0
backend/src/__init__.py Normal file
View file

82
backend/src/main.py Normal file
View file

@ -0,0 +1,82 @@
import argparse
import uvicorn
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from src.routes.projects import router as project_router
from src.routes.tags import router as tag_router
from src.routes.records import router as record_router
description = """
Welcome to the Juggl API.
This API allows you to retrieve your timetracking data.
"""
app = FastAPI(
title="Juggl API",
description=description,
contact={ # Contact information. See https://fastapi.tiangolo.com/tutorial/metadata/
"name": "Juggl Management",
"url": "https://juggl.giller.dev/contact",
"email": "help@juggl.giller.dev",
},
docs_url="/" # Documentation on start page
)
"""The main FastAPI instance"""
# Enable CORS to allow requests from other domains/origins.
# See https://en.wikipedia.org/wiki/Cross-origin_resource_sharing
app.add_middleware(
CORSMiddleware,
allow_origins=[
# Allow **all** foreign requests to access this API.
'*'
],
allow_methods=["*"],
allow_headers=["*"],
)
# Add subroutes
app.include_router(project_router)
app.include_router(tag_router)
app.include_router(record_router)
def main():
args = parse_args()
if args.dev:
start_dev_server(args.port)
else:
start_production_server(args.port)
def parse_args() -> argparse.Namespace:
"""Parses command line arguments on application startup."""
parser = argparse.ArgumentParser(
description="Juggl Backend v2",
formatter_class=argparse.ArgumentDefaultsHelpFormatter # show default values
)
parser.add_argument("--dev", default=False, action=argparse.BooleanOptionalAction,
help="Start a dev server with auto-reloading")
parser.add_argument("--port", type=int, default=8192, help="Port on which the API should run")
return parser.parse_args()
def start_dev_server(port: int):
"""Starts a development server with auto-reloading on port."""
uvicorn.run("main:app", reload=True, port=port)
def start_production_server(port: int):
"""Starts a production, i.e. performance-optimized server on port."""
uvicorn.run("main:app", port=port)
if __name__ == "__main__":
main()

View file

View file

@ -0,0 +1,25 @@
from datetime import datetime, timedelta
from pydantic import BaseModel, Field
from pydantic.json import timedelta_isoformat
class ProjectNew(BaseModel):
"""Model used when a user creates a new project."""
name: str = Field(description="Name of the project", example="Learning")
class ProjectInfo(BaseModel):
"""Model used when querying information about a module."""
name: str = Field(description="Name of the project", example="Learning")
start_date: datetime = Field(description="Project creation date", example=datetime.now())
duration: timedelta = Field(description="Total tracked duration", example=timedelta(days=3, hours=5, minutes=47))
records: int = Field(description="Total number of trackings/records", example=42)
class Config: # TODO: Adding this config may be done with a decorator
"""pydantic config"""
json_encoders = {
# Serialize timedeltas as ISO 8601 and not as float seconds, which is the default
timedelta: timedelta_isoformat,
}

View file

@ -0,0 +1,27 @@
from datetime import datetime, timedelta
from pydantic import BaseModel, Field
from pydantic.json import timedelta_isoformat
class RecordStart(BaseModel):
"""Model used when a user starts a record."""
start: datetime = Field(
description="Start date of record",
example=datetime.now() - timedelta(hours=3)
)
project: str = Field(description="Name of the project that should be tracked", example="Uni")
tags: list[str] = Field(description="List of tags by name", example=["Listening Lectures", "Not be concentrated"])
class RecordInfo(BaseModel):
start: datetime = Field(
description="Start date of record",
example=datetime.now() - timedelta(hours=3)
)
end: datetime = Field(
description="End date of record",
example=datetime.now()
)
project: str = Field(description="Name of the project that should be tracked", example="Uni")
tags: list[str] = Field(description="List of tags by name", example=["Listening Lectures", "Not be concentrated"])

16
backend/src/models/tag.py Normal file
View file

@ -0,0 +1,16 @@
from datetime import datetime, timedelta
from pydantic import BaseModel, Field
from pydantic.json import timedelta_isoformat
class TagNew(BaseModel):
"""Model used when a user creates a new tag."""
name: str = Field(description="Name of the tag", example="Listening Lectures")
class TagInfo(BaseModel):
"""Model used when querying information about a tag."""
name: str = Field(description="Name of the tag", example="Listening Lectures")
duration: timedelta = Field(description="Total tracked duration", example=timedelta(days=3, hours=5, minutes=47))

View file

View file

@ -0,0 +1,66 @@
import logging
from datetime import datetime, timedelta
from fastapi import APIRouter, status, Path
from src.models.project import ProjectNew, ProjectInfo
router = APIRouter(prefix='/projects', tags=["Projects"])
@router.get("/", response_model=list[ProjectInfo], summary="Get a list of all projects")
async def all_projects() -> list[ProjectInfo]:
"""Returns a list of all projects"""
return [
ProjectInfo(
name="Learning",
start_date=datetime.now(),
duration=timedelta(days=3, hours=5, minutes=47),
records=42
),
ProjectInfo(
name="Sports",
start_date=datetime.now(),
duration=timedelta(hours=14, minutes=10),
records=10
),
ProjectInfo(
name="Gaming",
start_date=datetime.now(),
duration=timedelta(hours=3, minutes=2),
records=5
)
]
@router.post("/", status_code=status.HTTP_201_CREATED, summary="Add a project")
async def add_project(project: ProjectNew):
"""Add a project."""
print('adding project', project)
@router.get("/{name}", response_model=ProjectInfo, summary="Get a project by name")
async def get_project(name: str = Path(..., title="Name of the module")) -> ProjectInfo:
"""Fetch a project by name."""
return ProjectInfo(
name=name,
start_date=datetime.now(),
duration=timedelta(hours=3, minutes=2),
records=5
)
@router.delete("/{name}", status_code=status.HTTP_204_NO_CONTENT, summary="Delete a project by name")
async def delete_project(name: str = Path(..., title="Name of the module")):
"""Delete a module specified by name."""
logging.debug(f"Deleting module {name}")
@router.patch("/{name}", summary="Apply partial updates to a project by name")
async def patch_project(project: ProjectNew, name: str = Path(..., title="Name of the module")):
"""Apply partial updates to a project."""
logging.debug(f"Patching project {name} with {project}")

View file

@ -0,0 +1,67 @@
import logging
from datetime import datetime, timedelta
from fastapi import APIRouter, status, Path
from src.models.record import RecordStart, RecordInfo
router = APIRouter(prefix="/records", tags=["Records"])
@router.get("/", response_model=list[RecordInfo], summary="Get a list of all records")
async def all_tags() -> list[RecordInfo]:
"""Returns a list of all records."""
return [
RecordInfo(
start=datetime.now() - timedelta(hours=3),
end=datetime.now(),
project="Uni",
tags=["Listening Lectures", "Not be concentrated"]
),
RecordInfo(
start=datetime.now() - timedelta(hours=7),
end=datetime.now() - timedelta(hours=4),
project="Uni",
tags=["Listening Lectures", "Not be concentrated"]
)
]
@router.post("/start", status_code=status.HTTP_201_CREATED, summary="Start a record")
async def start_record(record: RecordStart):
"""Start a record."""
print('starting record', record)
@router.post("/end", status_code=status.HTTP_204_NO_CONTENT, summary="End a record")
async def end_record(record: RecordStart):
"""End a record."""
print('end record', record)
@router.get("/{id}", response_model=RecordInfo, summary="Get a record by id")
async def get_record(id: str = Path(..., title="ID of the record")) -> RecordInfo:
"""Fetch a record by id."""
return RecordInfo(
start=datetime.now() - timedelta(hours=3),
end=datetime.now(),
project="Uni",
tags=["Listening Lectures", "Not be concentrated"]
)
@router.delete("/{id}", status_code=status.HTTP_204_NO_CONTENT, summary="Delete a record by id")
async def delete_record(id: str = Path(..., title="ID of the record")):
"""Delete a tag specified by name."""
logging.debug(f"Deleting record {id}")
@router.patch("/{id}", summary="Apply partial updates to a record by id")
async def patch_record(record: RecordInfo, id: str = Path(..., title="ID of the record")):
"""Apply partial updates to a record."""
logging.debug(f"Patching record {id} with {record}")

View file

@ -0,0 +1,54 @@
import logging
from datetime import datetime, timedelta
from fastapi import APIRouter, status, Path
from src.models.tag import TagNew, TagInfo
router = APIRouter(prefix="/tags", tags=["Tags"])
@router.get("/", response_model=list[TagInfo], summary="Get a list of all tags")
async def all_tags() -> list[TagInfo]:
"""Returns a list of all tags."""
return [
TagInfo(
name="Listening Lectures",
duration=timedelta(days=3, hours=5, minutes=47),
),
TagInfo(
name="Doing homework",
duration=timedelta(hours=14, minutes=10),
)
]
@router.post("/", status_code=status.HTTP_201_CREATED, summary="Add a tag")
async def add_tag(tag: TagNew):
"""Add a tag."""
print('adding tag', tag)
@router.get("/{name}", response_model=TagInfo, summary="Get a project by name")
async def get_tag(name: str = Path(..., title="Name of the tag")) -> TagInfo:
"""Fetch a project by name."""
return TagInfo(
name=name,
duration=timedelta(hours=3, minutes=2),
)
@router.delete("/{name}", status_code=status.HTTP_204_NO_CONTENT, summary="Delete a tag by name")
async def delete_tag(name: str = Path(..., title="Name of the tag")):
"""Delete a tag specified by name."""
logging.debug(f"Deleting tag {name}")
@router.patch("/{name}", summary="Apply partial updates to a tag by name")
async def patch_tag(tag: TagNew, name: str = Path(..., title="Name of the tag")):
"""Apply partial updates to a tag."""
logging.debug(f"Patching tag {name} with {tag}")