Basic examples with services
The documentation is under heavy development!
The documentation is actively being developed and might not cover every feature. For full details, refer to the source code.
No specification
The following example demonstrates how to create a basic service without a resource specification. This is useful when you want to create a service that doesn't need any specific resources. Unlike services with specifications, basic services can also be attached directly to packages.
Implementation
Create service
Create a service without a resource specification
from plateforme import BaseService, route
class NoSpecService(BaseService):
"""A service without specification."""
@route.post()
async def hello(self) -> str: # (1)!
return 'Hello, world!'
- We create a simple
POSTendpoint that returns a string message.
Assign the service
Assign the service to a resource
from plateforme import BaseResource, ConfigDict
class Material(BaseResource):
__config__ = ConfigDict(services=[NoSpecService])
Source code in src/examples/basic_services/example_no_spec.py
from pathlib import Path
from plateforme import Plateforme
dirname = Path(__file__).parent
# Create application
app = Plateforme(
debug=True,
logging={'level': 'DEBUG'},
database_engines=dirname / 'plateforme.db',
)
# Example
from plateforme import BaseService, route
class NoSpecService(BaseService):
"""A service without specification."""
@route.post()
async def hello(self) -> str: # (1)!
return 'Hello, world!'
from plateforme import BaseResource, ConfigDict
class Material(BaseResource):
__config__ = ConfigDict(services=[NoSpecService])
# Setup database
if __name__ == '__main__':
app.metadata.create_all()
Playground
Call hello endpoint
We can directly call the service hello endpoint.
With simple specification
The following example demonstrates how to create a basic service with a simple specification. This is useful when you want to create a service that can be attached to multiple resources. The specification is like a contract that the resources must fulfill to be able to use the service. In this example, we create a service that can be attached to "orderable" resources, i.e. resources that expose a price attribute.
Implementation
Create service
Create a service with a simple specification
from plateforme import BaseServiceWithSpec, BaseSpec
from plateforme.api import AsyncSessionDep, route
from plateforme.database import select
class OrderableSpec(BaseSpec):
"""An orderable specification."""
price: float
class SimpleSpecService(BaseServiceWithSpec[OrderableSpec]):
"""A service with a simple specification."""
@route.post()
async def buy(
self,
session: AsyncSessionDep,
payload: dict[str, int],
) -> float:
ids = [int(k) for k in payload.keys()]
query = select(self.resource).filter(self.resource.id.in_(ids))
buffer = await session.execute(query)
result = buffer.scalars().all()
if len(result) != len(payload):
raise ValueError('Invalid payload')
total = 0
for item in result:
total += item.price * payload[str(item.id)]
return total
Assign the service
Assign the service to some resources
from plateforme import ConfigDict, CRUDResource, Field
class Material(CRUDResource):
__config__ = ConfigDict(services=[..., SimpleSpecService])
code: str = Field(unique=True)
name: str
price: float
class Rocket(CRUDResource):
__config__ = ConfigDict(services=[..., SimpleSpecService])
code: str = Field(unique=True)
name: str
price: float
Source code in src/examples/basic_services/example_simple_spec.py
from pathlib import Path
from plateforme import Plateforme
dirname = Path(__file__).parent
# Create application
app = Plateforme(
debug=True,
logging={'level': 'DEBUG'},
database_engines=dirname / 'plateforme.db',
)
# Example
from plateforme import BaseServiceWithSpec, BaseSpec
from plateforme.api import AsyncSessionDep, route
from plateforme.database import select
class OrderableSpec(BaseSpec):
"""An orderable specification."""
price: float
class SimpleSpecService(BaseServiceWithSpec[OrderableSpec]):
"""A service with a simple specification."""
@route.post()
async def buy(
self,
session: AsyncSessionDep,
payload: dict[str, int],
) -> float:
ids = [int(k) for k in payload.keys()]
query = select(self.resource).filter(self.resource.id.in_(ids))
buffer = await session.execute(query)
result = buffer.scalars().all()
if len(result) != len(payload):
raise ValueError('Invalid payload')
total = 0
for item in result:
total += item.price * payload[str(item.id)]
return total
from plateforme import ConfigDict, CRUDResource, Field
class Material(CRUDResource):
__config__ = ConfigDict(services=[..., SimpleSpecService])
code: str = Field(unique=True)
name: str
price: float
class Rocket(CRUDResource):
__config__ = ConfigDict(services=[..., SimpleSpecService])
code: str = Field(unique=True)
name: str
price: float
# Setup database
if __name__ == '__main__':
app.metadata.create_all()
Playground
Add some resources
Let's create some materials and rockets.
Add some materials
POST http://127.0.0.1:8001/materials HTTP/1.1
content-type: application/json
{
"payload": [
{
"name": "Main Engine",
"code": "ME-1000",
"price": 1500000.00
},
{
"name": "Primary Fuel Tank",
"code": "FT-500",
"price": 145000.00
},
{
"name": "Guidance System",
"code": "GS-200",
"price": 120000.00
}
]
}
Response
Add some rockets
POST http://127.0.0.1:8001/rockets HTTP/1.1
content-type: application/json
{
"payload": [
{
"name": "Saturn V",
"code": "SAT-V",
"price": 240000000.00
},
{
"name": "Falcon 9",
"code": "FAL-9",
"price": 62000000.00
},
{
"name": "Space Launch System",
"code": "SLS-1",
"price": 500000000.00
}
]
}
Call buy endpoint
We can now call the service buy endpoint on materials and rockets.
With selection specification
The following example demonstrates how to create a basic service with a selection specification. This is useful when you want to create endpoints either at the resource collection level or at the resource instance level. The selection parameter will be filled at runtime with the selected resources inferred from the URL. In this example, we create a service that can be attached to both materials and rockets.
Implementation
Create service
Create a service with a selection specification
from plateforme import BaseServiceWithSpec, BaseSpec, Key, KeyList
from plateforme.api import AsyncSessionDep, Selection, route
class DescriptionSpec(BaseSpec):
"""A description specification."""
code: str
name: str
class SelectionSpecService(BaseServiceWithSpec[DescriptionSpec]):
"""A service with a selection specification."""
@route.get(path='/describe')
async def describe_one(
self,
session: AsyncSessionDep,
selection: Key[DescriptionSpec] = Selection(),
) -> str:
result = await selection.resolve(session)
return f"{result.name} ({result.code})"
@route.get(path='/describe')
async def describe_many(
self,
session: AsyncSessionDep,
selection: KeyList[DescriptionSpec] = Selection(),
) -> list[str]:
result = await selection.resolve(session)
descriptions = []
for item in result:
descriptions.append(f"{item.name} ({item.code})")
return descriptions
Assign the service
Assign the service to some resources
from plateforme import ConfigDict, CRUDResource, Field
class Material(CRUDResource):
__config__ = ConfigDict(services=[..., SelectionSpecService])
code: str = Field(unique=True)
name: str
class Rocket(CRUDResource):
__config__ = ConfigDict(services=[..., SelectionSpecService])
code: str = Field(unique=True)
name: str
Source code in src/examples/basic_services/example_selection_spec.py
from pathlib import Path
from plateforme import Plateforme
dirname = Path(__file__).parent
# Create application
app = Plateforme(
debug=True,
logging={'level': 'DEBUG'},
database_engines=dirname / 'plateforme.db',
)
# Example
from plateforme import BaseServiceWithSpec, BaseSpec, Key, KeyList
from plateforme.api import AsyncSessionDep, Selection, route
class DescriptionSpec(BaseSpec):
"""A description specification."""
code: str
name: str
class SelectionSpecService(BaseServiceWithSpec[DescriptionSpec]):
"""A service with a selection specification."""
@route.get(path='/describe')
async def describe_one(
self,
session: AsyncSessionDep,
selection: Key[DescriptionSpec] = Selection(),
) -> str:
result = await selection.resolve(session)
return f"{result.name} ({result.code})"
@route.get(path='/describe')
async def describe_many(
self,
session: AsyncSessionDep,
selection: KeyList[DescriptionSpec] = Selection(),
) -> list[str]:
result = await selection.resolve(session)
descriptions = []
for item in result:
descriptions.append(f"{item.name} ({item.code})")
return descriptions
from plateforme import ConfigDict, CRUDResource, Field
class Material(CRUDResource):
__config__ = ConfigDict(services=[..., SelectionSpecService])
code: str = Field(unique=True)
name: str
class Rocket(CRUDResource):
__config__ = ConfigDict(services=[..., SelectionSpecService])
code: str = Field(unique=True)
name: str
# Setup database
if __name__ == '__main__':
app.metadata.create_all()
Playground
Add some resources
Let's create some materials and rockets.
Call describe-many endpoint
We can now call the service describe endpoint on materials and rockets.
Call describe-one endpoint
Or, alternatively, we can call the service describe endpoint on a specific material or rocket.