Plugins for FastAPI framework
Project description
Plugins for FastAPI framework, high performance, easy to learn, fast to code, ready for production
fastapi-plugins
FastAPI framework plugins - simple way to share fastapi
code and utilities across applications.
The concept is plugin
- plug a functional utility into your application without or with minimal effort.
- Cache
- Scheduler
- Control
- Application settings/configuration
- Celery
- MQ
- and much more is already in progress...
Changes
See release notes
Installation
pip install fastapi-plugins
pip install fastapi-plugins[memcached]
pip install fastapi-plugins[all]
Quick start
Plugin
Add information about plugin system.
Application settings
Add information about settings.
Application configuration
Add information about configuration of an application
Complete example
import fastapi
import fastapi_plugins
from fastapi_plugins.memcached import MemcachedSettings
from fastapi_plugins.memcached import MemcachedClient
from fastapi_plugins.memcached import depends_memcached
from fastapi_plugins.memcached import memcached_plugin
import asyncio
import aiojobs
import aioredis
@fastapi_plugins.registered_configuration
class AppSettings(
fastapi_plugins.ControlSettings,
fastapi_plugins.RedisSettings,
fastapi_plugins.SchedulerSettings,
MemcachedSettings,
):
api_name: str = str(__name__)
@fastapi_plugins.registered_configuration(name='sentinel')
class AppSettingsSentinel(AppSettings):
redis_type = fastapi_plugins.RedisType.sentinel
redis_sentinels = 'localhost:26379'
app = fastapi.FastAPI()
config = fastapi_plugins.get_config()
@app.get("/")
async def root_get(
cache: aioredis.Redis=fastapi.Depends(fastapi_plugins.depends_redis),
conf: pydantic.BaseSettings=fastapi.Depends(fastapi_plugins.depends_config) # noqa E501
) -> typing.Dict:
return dict(ping=await cache.ping(), api_name=conf.api_name)
@app.post("/jobs/schedule/<timeout>")
async def job_post(
timeout: int=fastapi.Query(..., title='the job sleep time'),
cache: aioredis.Redis=fastapi.Depends(fastapi_plugins.depends_redis),
scheduler: aiojobs.Scheduler=fastapi.Depends(fastapi_plugins.depends_scheduler), # noqa E501
) -> str:
async def coro(job_id, timeout, cache):
await cache.set(job_id, 'processing')
try:
await asyncio.sleep(timeout)
if timeout == 8:
raise Exception('ugly error')
except asyncio.CancelledError:
await cache.set(job_id, 'canceled')
except Exception:
await cache.set(job_id, 'erred')
else:
await cache.set(job_id, 'success')
job_id = str(uuid.uuid4()).replace('-', '')
await cache.set(job_id, 'pending')
await scheduler.spawn(coro(job_id, timeout, cache))
return job_id
@app.get("/jobs/status/<job_id>")
async def job_get(
job_id: str=fastapi.Query(..., title='the job id'),
cache: aioredis.Redis=fastapi.Depends(fastapi_plugins.depends_redis),
) -> typing.Dict:
status = await cache.get(job_id)
if status is None:
raise fastapi.HTTPException(
status_code=starlette.status.HTTP_404_NOT_FOUND,
detail='Job %s not found' % job_id
)
return dict(job_id=job_id, status=status)
@app.post("/memcached/demo/<key>")
async def memcached_demo_post(
key: str=fastapi.Query(..., title='the job id'),
cache: MemcachedClient=fastapi.Depends(depends_memcached),
) -> typing.Dict:
await cache.set(key.encode(), str(key + '_value').encode())
value = await cache.get(key.encode())
return dict(ping=(await cache.ping()).decode(), key=key, value=value)
@app.on_event('startup')
async def on_startup() -> None:
await fastapi_plugins.config_plugin.init_app(app, config)
await fastapi_plugins.config_plugin.init()
await memcached_plugin.init_app(app, config)
await memcached_plugin.init()
await fastapi_plugins.redis_plugin.init_app(app, config=config)
await fastapi_plugins.redis_plugin.init()
await fastapi_plugins.scheduler_plugin.init_app(app=app, config=config)
await fastapi_plugins.scheduler_plugin.init()
await fastapi_plugins.control_plugin.init_app(
app,
config=config,
version=__version__,
environ=config.dict()
)
await fastapi_plugins.control_plugin.init()
@app.on_event('shutdown')
async def on_shutdown() -> None:
await fastapi_plugins.control_plugin.terminate()
await fastapi_plugins.scheduler_plugin.terminate()
await fastapi_plugins.redis_plugin.terminate()
await memcached_plugin.terminate()
await fastapi_plugins.config_plugin.terminate()
Development
Issues and suggestions are welcome through issues
License
This project is licensed under the terms of the MIT license.
Changes
0.8.1 (2020-03-31)
[fix]
Fix settings for Python 3.7
0.8.0 (2020-03-31)
[feature]
Settings plugin
0.7.0 (2020-03-29)
[feature]
Control plugin with Health, Heartbeat, Environment and Version
0.6.1 (2020-03-24)
[fix]
Bumpaiojobs
to get rid of not required dependencies
0.6.0 (2020-11-26)
[feature]
Memcached
0.5.0 (2020-11-25)
- [bug] remove
__all__
since no API as such (#6). - [typo] Fix typos in README (#7).
- [feature] Add Redis TTL (#8).
0.4.2 (2020-11-24)
- [bug] Fix Redis URL (#4).
0.4.1 (2020-06-16)
- Refactor requirements
0.4.0 (2020-04-09)
- structure and split dependencies to
extra
0.3.0 (2020-04-07)
- Scheduler: tasks scheduler based on
aiojobs
0.2.1 (2020-04-06)
- Redis: pre-start
0.2.0 (2019-12-11)
- Redis: sentinels
0.1.0 (2019-11-20)
- Initial release: simple redis pool client
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
fastapi-plugins-0.8.1.tar.gz
(17.9 kB
view hashes)
Built Distribution
Close
Hashes for fastapi_plugins-0.8.1-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | c57b7328068f615ec6e964aefacd9935c53ad9406b4a5a0fc8068a60c83c430d |
|
MD5 | 224a9a0730bcd8b4935959209143bd31 |
|
BLAKE2b-256 | b38279d33aeca125ab52db7741de5f9e65bbdc99b88d3ff8d378f01b166e4bd9 |