Skip to content

Example

Simple Demo⚓︎

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
import uvicorn
from pydantic import BaseModel
from starlette.responses import JSONResponse

from star_openapi import OpenAPI

info = {"title": "Star API", "version": "1.0.0"}
app = OpenAPI(info=info)

book_tag = {"name": "book", "description": "book tag"}


class BookModel(BaseModel):
    name: str
    age: int


@app.get("/book", summary="get books", tags=[book_tag])
async def get_book(query: BookModel):
    """
    get all books
    """
    print(query.model_dump_json())
    return JSONResponse({"message": "Hello World"})


if __name__ == "__main__":
    print(app.routes)
    uvicorn.run(app)

REST Demo⚓︎

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
from http import HTTPStatus

from pydantic import BaseModel, Field
from starlette.responses import JSONResponse

from star_openapi import Info, OpenAPI, Tag

info = Info(title="book API", version="1.0.0")
# Basic Authentication Sample
basic = {"type": "http", "scheme": "basic"}
# JWT Bearer Sample
jwt = {"type": "http", "scheme": "bearer", "bearerFormat": "JWT"}
# API Key Sample
api_key = {"type": "apiKey", "name": "api_key", "in": "header"}
# Implicit OAuth2 Sample
oauth2 = {
    "type": "oauth2",
    "flows": {
        "implicit": {
            "authorizationUrl": "https://example.com/api/oauth/dialog",
            "scopes": {"write:pets": "modify pets in your account", "read:pets": "read your pets"},
        }
    },
}
security_schemes = {"jwt": jwt, "api_key": api_key, "oauth2": oauth2, "basic": basic}


class NotFoundResponse(BaseModel):
    code: int = Field(-1, description="Status Code")
    message: str = Field("Resource not found!", description="Exception Information")


app = OpenAPI(info=info, security_schemes=security_schemes)

book_tag = Tag(name="book", description="Some Book")
security = [{"jwt": []}, {"oauth2": ["write:pets", "read:pets"]}]


class BookPath(BaseModel):
    id: int = Field(..., description="book id")


class BookQuery(BaseModel):
    age: int | None = Field(None, description="Age")
    s_list: list[str] = Field(None, alias="s_list[]", description="some array")


class BookBody(BaseModel):
    age: int | None = Field(..., ge=2, le=4, description="Age")
    author: str | None = Field(None, min_length=2, max_length=4, description="Author")


class BookBodyWithID(BaseModel):
    id: int = Field(..., description="book id")
    age: int | None = Field(None, ge=2, le=4, description="Age")
    author: str = Field(None, min_length=2, max_length=4, description="Author")


class BookResponse(BaseModel):
    code: int = Field(0, description="Status Code")
    message: str = Field("ok", description="Exception Information")
    data: BookBodyWithID | None


@app.get(
    "/book/{id}",
    tags=[book_tag],
    summary="new summary",
    description="new description",
    security=security,
)
async def get_book(path: BookPath):
    """Get a book
    to get some book by id, like:
    http://localhost:8000/book/3
    """
    if path.id == 4:
        return NotFoundResponse().model_dump_json(), 404
    return JSONResponse({"code": 0, "message": "ok", "data": {"id": path.id, "age": 3, "author": "no"}})


# set doc_ui False disable openapi UI
@app.get("/book", doc_ui=True, deprecated=True)
async def get_books(query: BookQuery):
    """get books
    to get all books
    """
    print(query)
    return JSONResponse(
        {
            "code": 0,
            "message": "ok",
            "data": [{"id": 1, "age": query.age, "author": "a1"}, {"id": 2, "age": query.age, "author": "a2"}],
        }
    )


@app.post("/book", tags=[book_tag])
async def create_book(body: BookBody):
    print(body)
    return JSONResponse({"code": 0, "message": "ok"})


@app.put("/book/{id}", tags=[book_tag])
async def update_book(path: BookPath, body: BookBody):
    print(path)
    print(body)
    return JSONResponse({"code": 0, "message": "ok"})


@app.delete("/book/{id}", tags=[book_tag], doc_ui=False)
async def delete_book(path: BookPath):
    print(path)
    return JSONResponse({"code": 0, "message": "ok"})


if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app)

APIRouter⚓︎

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
import uvicorn
from pydantic import BaseModel
from starlette.responses import JSONResponse

from star_openapi import OpenAPI
from star_openapi.router import APIRouter

app = OpenAPI()

api1 = APIRouter(url_prefix="/api1")
api2 = APIRouter(url_prefix="/api2")


class IdModel(BaseModel):
    id: int


class BookModel(BaseModel):
    name: str
    age: int


@api1.get("/book")
async def ge_book():
    return JSONResponse({"message": "Hello World1"})


@api1.post("/book")
async def create_book(body: BookModel):
    return JSONResponse(body.model_dump())


@api1.put("/book/{id}")
async def update_book(path: IdModel, body: BookModel):
    return JSONResponse({"id": path.id, "name": body.name, "age": body.age})


@api1.delete("/book/{id}")
async def delete_book(path: IdModel):
    return JSONResponse({"id": path.id})


@api2.get("/")
async def get_api2():
    return JSONResponse({"message": "Hello World2"})


api1.register_api(api2)

app.register_api(api1)
app.register_api(api2)

if __name__ == "__main__":
    print(app.routes)
    uvicorn.run(app)

Upload File Demo⚓︎

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
import uvicorn
from pydantic import BaseModel, Field
from starlette.responses import JSONResponse

from star_openapi import OpenAPI, UploadFile

app = OpenAPI()


class UploadFileForm(BaseModel):
    file: UploadFile
    file_type: str = Field(None, description="File Type")


class UploadFilesForm(BaseModel):
    files: list[UploadFile]
    file_type: str = Field(None, description="File Type")


@app.post("/upload/file")
async def upload_file(form: UploadFileForm):
    print(form.file.filename)
    print(form.file_type)

    content = await form.file.read()
    with open(form.file.filename, "wb") as f:
        f.write(content)
    return JSONResponse({"code": 0, "message": "ok"})


@app.post("/upload/files")
async def upload_files(form: UploadFilesForm):
    print(form.files)
    print(form.file_type)
    return JSONResponse({"code": 0, "message": "ok"})


if __name__ == "__main__":
    print(app.routes)
    uvicorn.run(app)

Websocket Demo⚓︎

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
import uvicorn
from starlette.websockets import WebSocket

from star_openapi import OpenAPI

app = OpenAPI()


@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    while True:
        data = await websocket.receive_text()
        await websocket.send_text(data)


if __name__ == "__main__":
    uvicorn.run("websocket:app", reload=True)

A complete project⚓︎

see star-api-demo