Extension for starlette endpoint classes with the following goals:
- Generic endpoints
- Implement a flexibly customisable and easy to use generic endpoint as a high level abstraction on top of the
starlette.endpoints.HTTPEndpoint - Provide a set of higher level endpoints that implements common cases, eg. CollectionEndpoint, ItemEndpoint
- Provide an interface to easily customise the class by users needs
- Implement a flexibly customisable and easy to use generic endpoint as a high level abstraction on top of the
- Schemas backends (deserialization, validation, serialization)
- Implement a generic interface for different schema backends bootstrapping
- Provide a set of common schema backend implementations:
pydantic,typesystem,marshmallow - Provide an interface for the user custom implementations
- Schema generation backends
- Extend the existing (
starlette.schemas.SchemaGenerator) utilising the schema backend and generic endpoints implementation to automatically generate the API schema - Implement
Open API v3automatic schema generator backend that complies with the specs - Provide an interface for the user custom implementations
- Extend the existing (
- (Possibly) DB Models interface
- Implement a generic model interface to allow generalising common DB operations in the related endpoints utilising the schema details, so common endpoints will require only schemas implementations, eg:
- pagination fetch for
CollectionEndpoint.getor item creation inCollectionEndpoint.post - item read/update/delete as
ItemEndpoint.get,ItemEndpoint.putandItemEndpoint.deleteaccordingly
- pagination fetch for
- Implement a generic model interface to allow generalising common DB operations in the related endpoints utilising the schema details, so common endpoints will require only schemas implementations, eg:
The current state of the library is the proof of concept (POC)
TBD
Endpoint and schemas definitions with pydantic
authors.py
import typing
import aiosqlite
from starlette_cbge.endpoints import PydanticBaseEndpoint
from starlette_cbge.schema_backends import PydanticSchema, PydanticListSchema
from example_app.db import database
class AuthorGetCoolectionRequestSchema(PydanticSchema):
limit: int = 100
offset: int = 0
class AuthorPostRequestSchema(PydanticSchema):
name: str
class AuthorResponseSchema(PydanticSchema):
id: int
name: str
class AuthorResponseListSchema(PydanticListSchema):
id: int
name: str
class Authors(PydanticBaseEndpoint):
"""
Collection endpoint.
"""
request_schemas = (
("GET", AuthorGetCoolectionRequestSchema),
("POST", AuthorPostRequestSchema),
)
response_schemas = (
("GET", AuthorResponseListSchema),
("POST", AuthorResponseSchema),
)
async def get(self, request_data: typing.Dict) -> typing.List[aiosqlite.Row]:
"""
Retrieves the list of authors.
List is limited with `limit` and `offset` fields.
"""
async with database.connection() as connection:
raw_connection = connection.raw_connection
raw_connection.row_factory = aiosqlite.Row
query = """SELECT * FROM authors LIMIT :limit OFFSET :offset"""
cursor = await raw_connection.execute(query, request_data)
return await cursor.fetchall()
async def post(self, request_data: typing.Dict) -> aiosqlite.Row:
"""
Creates a new author and returns the created record
"""
async with database.connection() as connection:
raw_connection = connection.raw_connection
raw_connection.row_factory = aiosqlite.Row
query = "INSERT INTO authors (name) VALUES (:name);"
await raw_connection.execute(query, request_data)
query = "SELECT * FROM authors where id = last_insert_rowid();"
cursor = await raw_connection.execute(query)
return await cursor.fetchone()or for typesystem
from starlette_cbge.endpoints import TypesystemBaseEndpoint
from starlette_cbge.schema_backends import (
TypesystemSchema,
TypesystemListSchema,
typesystem_fields,
)
class AuthorGetCoolectionRequestSchema(TypesystemSchema):
limit = typesystem_fields.Integer(default=100)
offset = typesystem_fields.Integer(default=0)
class AuthorPostRequestSchema(TypesystemSchema):
name = typesystem_fields.String()
class AuthorResponseSchema(TypesystemSchema):
id = typesystem_fields.Integer()
name = typesystem_fields.String()
class AuthorResponseListSchema(TypesystemListSchema):
id = typesystem_fields.Integer()
name = typesystem_fields.String()
class Authors(TypesystemBaseEndpoint):
"""
Collection endpoint.
"""
request_schemas = (
("GET", AuthorGetCoolectionRequestSchema),
("POST", AuthorPostRequestSchema),
)
response_schemas = (
("GET", AuthorResponseListSchema),
("POST", AuthorResponseSchema),
)
# Same methods implementationsSchema generation
from starlette_cbge.schema_generator_backends import OpenAPIv3SchemaGenerator
from example_app.app import app
schemas = OpenAPIv3SchemaGenerator(
{"openapi": "3.0.0", "info": {"title": "Example API", "version": "1.0"}}
)
schema = schemas.get_schema(routes=app.routes)Generated schema
NOTE: it's not finalised output
{
"openapi":"3.0.0",
"info":{
"title":"Example API",
"version":"1.0"
},
"paths":{
"/authors":{
"get":{
"description":"Retrieves the list of authors. List is limited with `limit` and `offset` fields.",
"parameters":{
"title":"AuthorGetCoolectionRequestSchema",
"type":"object",
"properties":{
"limit":{
"title":"Limit",
"default":100,
"type":"integer"
},
"offset":{
"title":"Offset",
"default":0,
"type":"integer"
}
}
},
"responses":{
"200":{
"description":"Successful response",
"content":{
"application/json":{
"schema":{
"title":"AuthorResponseListSchema",
"type":"object",
"properties":{
"id":{
"title":"Id",
"type":"integer"
},
"name":{
"title":"Name",
"type":"string"
}
},
"required":[
"id",
"name"
]
}
}
}
},
"422":{
"description":"Invalid request",
"content":{
"application/json":{
"schema":{
"title":"type",
"type":"object",
"properties":{
"detail":{
"title":"Detail",
"type":"string"
},
"errors":{
"title":"Errors",
"default":[
],
"type":"array",
"items":{
"type":"object"
}
}
},
"required":[
"detail"
]
}
}
}
}
}
},
"post":{
"description":"Creates a new authors and returns the created record",
"parameters":{
"title":"AuthorPostRequestSchema",
"type":"object",
"properties":{
"name":{
"title":"Name",
"type":"string"
}
},
"required":[
"name"
]
},
"responses":{
"200":{
"description":"Successful response",
"content":{
"application/json":{
"schema":{
"title":"AuthorResponseSchema",
"type":"object",
"properties":{
"id":{
"title":"Id",
"type":"integer"
},
"name":{
"title":"Name",
"type":"string"
}
},
"required":[
"id",
"name"
]
}
}
}
},
"422":{
"description":"Invalid request",
"content":{
"application/json":{
"schema":{
"title":"type",
"type":"object",
"properties":{
"detail":{
"title":"Detail",
"type":"string"
},
"errors":{
"title":"Errors",
"default":[
],
"type":"array",
"items":{
"type":"object"
}
}
},
"required":[
"detail"
]
}
}
}
}
}
}
}
}
}