Commit 9ddeea42 authored by Vlad Calin's avatar Vlad Calin

Basic structure

parents
.idea/
*.pyc
__pycache__/
*.egg-info/
\ No newline at end of file
include README.md
\ No newline at end of file
# workloads
A python framework for building work load processing scripts.
from workloads import params
from workloads.registry import WorkloadRegistry
from workloads.workload import Workload, step
registry = WorkloadRegistry()
@registry.register
class FileProcessingWorkload(Workload):
name = 'file_processing'
parameters = [
params.Parameter('test_1'),
params.Parameter('test_2'),
params.Parameter('test_3'),
]
@step('start')
def start(self):
pass
if __name__ == '__main__':
registry.cli()
from setuptools import setup, find_packages
def read_file(filename):
with open(filename, 'r') as f:
return f.read()
setup(
name='workloads',
version='0.0.1',
description='A python framework for building work load processing scripts.',
description_long=read_file('README.md'),
author='Vlad Călin',
author_email='vlad.s.calin@gmail.com',
packages=find_packages(exclude=['examples']),
install_requires=[
'click',
'celery',
'redis'
],
extras_require={
'dev': [
'pytest',
'sphinx',
'twine',
]
},
entry_points={
'console_scripts': [
'workloads = workloads.cli:cli'
]
}
)
import click
@click.group()
def cli():
pass
class Parameter(object):
def __init__(self, name):
self.name = name
import functools
import click
from workloads.workload import run_workload
def parameter_to_click_option(parameter):
return click.option("--" + parameter.name)
class WorkloadRegistry(object):
def __init__(self):
self.workloads = []
def register(self, workload_cls, name=None):
name = name or workload_cls.name
if not name:
raise ValueError('Workload name must be specified')
self.workloads.append(workload_cls)
def list(self):
return self.workloads
def get(self, name):
for wk in self.workloads:
if wk.name == name:
return wk
raise KeyError('No workload {} found'.format(name))
def get_cli(self):
@click.group()
def cli():
pass
@cli.command('list')
def list_workflows():
for wf in self.list():
print(wf.name)
@cli.group('run')
def run_cli():
pass
for wl in self.workloads:
decorator = run_cli.command(wl.name)
decorator = decorator(functools.partial(run_workload, wl))
decorator = click.option('--queue', required=True)(decorator)
for param in wl.parameters:
decorator = parameter_to_click_option(param)(decorator)
return cli
def cli(self):
return self.get_cli()()
class Workload(object):
name = None
parameters = []
def step(func):
pass
def run_workload(cls, queue, **params):
print(cls)
print(queue)
print(params)
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment