ramjet

Scripts Manager


Keywords
aiohttp
License
MIT
Install
pip install ramjet==1.8.1

Documentation

Ramjet

versions PyPI version versions Commitizen friendly

わが征くは星の大海 —— Yang Wen-li

后台脚本的引擎

Links

Install & Run

Need Python3.5.x.

# Install from pypi

$ pip install ramjet
# Install from source

$ python setup.py install
$ python -m ramjet [--debug=true]

Description

基于 asyncio 和 consurrent.futures 运行脚本(tasks)。

每一个 task 都需要实现接口 bind_task()

利用 ioloopthread_executorprocess_executor 自行实现运行逻辑。

Demo

异步

import random
import asyncio

from ramjet.engines import process_executor, thread_executor, ioloop


def bind_task():
    # 将任务添加进事件循环中
    asyncio.ensure_future(async_task())


async def async_task():
    await asyncio.sleep(3)
    for i in range(10):
        asyncio.ensure_future(async_child_task(i))


async def async_child_task(n):
    await asyncio.sleep(random.random())
    print('child task {} ok!'.format(n))

多线程 & 多进程

需要注意子进程没法回收,所以请确保 task 可以很好的结束。

from ramjet.engines import process_executor, thread_executor, ioloop


def bind_task():
    # 多线程
    thread_executor.submit(task, your_arguments)
    # 多进程
    process_executor.submit(task, your_arguments)


def task(*args, **kw):
    pass

定时任务

from ramjet.engines import process_executor, thread_executor, ioloop


def bind_task():
    delay = 3600
    ioloop.call_later(delay, task, your_auguments)


def task(*args, **kw):
    # 可以在 task 内设置下一次执行的时间
    # ioloop.call_later(delay, task, *args, **kw)

HTTP

from aiohttp import web

from ramjet.settings import logger


logger = logger.getChild('tasks.web_demo')


def bind_task():
    logger.info("run web_demo")


def bind_handle(add_route):
    add_route('/', DemoHandle)


class DemoHandle(web.View):

    async def get(self):
        return web.Response(text="New hope")

Versions

更新日志