Commit eba8f0d4 authored by Mark Steadman's avatar Mark Steadman

Initial commit

parents
Pipeline #3802 passed with stages
in 1 minute and 31 seconds
File added
# http://editorconfig.org
root = true
[*]
charset = utf-8
end_of_line = lf
insert_final_newline = true
trim_trailing_whitespace = true
[*.{py,rst,ini}]
indent_style = space
indent_size = 4
[*.{html,css,scss,json,yml}]
indent_style = space
indent_size = 2
[*.md]
trim_trailing_whitespace = false
[Makefile]
indent_style = tab
*.py[cod]
__pycache__
.env
# C extensions
*.so
# Packages
*.egg
*.egg-info
dist
build
eggs
parts
bin
include
var
sdist
develop-eggs
.installed.cfg
lib
lib64
# Installer logs
pip-log.txt
# Unit test / coverage reports
.coverage
.tox
nosetests.xml
htmlcov
# Translations
*.mo
# Mr Developer
.mr.developer.cfg
.project
.pydevproject
# Pycharm/Intellij
.idea
# Complexity
output/*.html
output/*/index.html
# Sphinx
docs/_build
stages:
- qa
- test
- build
flake8:
stage: qa
image: python:3.6.3
before_script:
- pip install -e git+https://gitlab.com/pycqa/flake8@9631dac5#egg=flake8
script:
- flake8 ./cron
test_django1:
stage: test
image: python:3.6.3
before_script:
- pip install -qr requirements.txt
- pip install -q "Django<2.0" coverage mock
script:
- coverage run --source cron runtests.py && coverage report
test_django2:
stage: test
image: python:3.6.3
before_script:
- pip install -qr requirements.txt
- pip install -q django coverage mock
script:
- coverage run --source cron runtests.py && coverage report
build:
stage: build
image: python:3.6.3
before_script:
- pip install -qU twine setuptools wheel
script:
- python setup.py sdist
- twine upload dist/* -u ${PYPI_USERNAME} -p ${PYPI_PASSWORD} --skip-existing
MIT License
Copyright (c) 2018, Mark Steadman
Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
include AUTHORS.rst
include CONTRIBUTING.rst
include HISTORY.rst
include LICENSE
include README.md
recursive-include cron *.html *.png *.gif *js *.css *jpg *jpeg *svg *py
Podiant cron
============
![Build](https://git.steadman.io/podiant/cron/badges/master/build.svg)
![Coverage](https://git.steadman.io/podiant/cron/badges/master/coverage.svg)
A tiny wrapper around APScheduler and Django RQ, for defining cron jobs
## Quickstart
Install cron:
```sh
pip install podiant-cron
```
Add it to your `INSTALLED_APPS`:
```python
INSTALLED_APPS = (
...
'cron',
...
)
```
## Running tests
Does the code actually work?
```
coverage run --source cron runtests.py
```
## Credits
Tools used in rendering this package:
- [Cookiecutter](https://github.com/audreyr/cookiecutter)
- [`cookiecutter-djangopackage`](https://github.com/pydanny/cookiecutter-djangopackage)
from apscheduler.schedulers.blocking import BlockingScheduler
from django_rq import enqueue
from django.conf import settings
from imp import find_module
from importlib import import_module
__version__ = '0.1.0'
scheduler = BlockingScheduler()
def interval(**kwargs):
def wrapper(f):
scheduler.add_job(
func=enqueue,
args=[f],
trigger='interval',
**kwargs
)
return f
return wrapper
def scheduled(**kwargs):
def wrapper(f):
scheduler.add_job(
func=enqueue,
args=[f],
trigger='cron',
**kwargs
)
return f
return wrapper
def autodiscover():
for app in settings.INSTALLED_APPS:
import_module(app)
name = '%s.cron' % app
try:
import_module(name)
except ImportError as ex:
try:
find_module(name)
except ImportError:
continue
raise ex
from django.core.management.base import BaseCommand
from ... import scheduler, autodiscover
import logging
class Command(BaseCommand):
help = 'Run scheduled jobs'
def handle(self, *args, **options):
autodiscover()
logger = logging.getLogger('podiant.cron')
logger.debug('Starting clock')
scheduler.start()
This diff is collapsed.
#!/usr/bin/env python
# -*- coding: utf-8 -*-
from __future__ import unicode_literals, absolute_import
import os
import sys
if __name__ == "__main__":
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "tests.settings")
from django.core.management import execute_from_command_line
execute_from_command_line(sys.argv)
{"last_check":"2018-10-03T16:06:39Z","pypi_version":"18.0"}
\ No newline at end of file
home = /usr/local/Cellar/python/3.7.0/libexec/bin
include-system-site-packages = false
version = 3.7.0
APScheduler==3.5.3
django-rq==1.2.0
rq-scheduler==0.7.0
rq==0.12
# Additional requirements go here
#!/usr/bin/env python
# -*- coding: utf-8
from __future__ import unicode_literals, absolute_import
import os
import sys
import django
from django.conf import settings
from django.test.utils import get_runner
def run_tests(*test_args):
if not test_args:
test_args = ['tests']
os.environ['DJANGO_SETTINGS_MODULE'] = 'tests.settings'
django.setup()
TestRunner = get_runner(settings)
test_runner = TestRunner()
failures = test_runner.run_tests(test_args)
sys.exit(bool(failures))
if __name__ == '__main__':
run_tests(*sys.argv[1:])
[bumpversion]
current_version = 0.1.0
commit = True
tag = True
[bumpversion:file:cron/__init__.py]
[wheel]
universal = 1
[flake8]
ignore = D203
exclude =
cron/migrations,
.git,
.tox,
docs/conf.py,
build,
dist
max-line-length = 119
[coverage:report]
omit=*/management/commands/*.py,*/migrations/*.py
#!/usr/bin/env python
# -*- coding: utf-8 -*-
import os
import re
import sys
try:
from setuptools import setup
except ImportError:
from distutils.core import setup
def get_version(*file_paths):
"""Retrieves the version from cron/__init__.py"""
filename = os.path.join(os.path.dirname(__file__), *file_paths)
version_file = open(filename).read()
version_match = re.search(r"^__version__ = ['\"]([^'\"]*)['\"]",
version_file, re.M)
if version_match:
return version_match.group(1)
raise RuntimeError('Unable to find version string.')
version = get_version("cron", "__init__.py")
if sys.argv[-1] == 'publish':
try:
import wheel
print("Wheel version: ", wheel.__version__)
except ImportError:
print('Wheel library missing. Please run "pip install wheel"')
sys.exit()
os.system('python setup.py sdist upload')
os.system('python setup.py bdist_wheel upload')
sys.exit()
if sys.argv[-1] == 'tag':
print("Tagging the version on git:")
os.system("git tag -a %s -m 'version %s'" % (version, version))
os.system("git push --tags")
sys.exit()
readme = open('README.md').read()
setup(
name='podiant-cron',
version=version,
description='A simpler, more flexible form of Django signals',
long_description=readme,
long_description_content_type='text/markdown',
author='Mark Steadman',
author_email='mark@podiant.co',
url='https://git.steadman.io/podiant/cron',
packages=['cron'],
include_package_data=True,
install_requires=[
'APScheduler',
'django-rq',
'rq-scheduler',
'rq'
],
license="MIT",
zip_safe=False,
keywords='podiant-cron',
classifiers=[
'Development Status :: 3 - Alpha',
'Framework :: Django :: 1.11',
'Intended Audience :: Developers',
'License :: OSI Approved :: BSD License',
'Natural Language :: English',
'Programming Language :: Python :: 3',
'Programming Language :: Python :: 3.4',
'Programming Language :: Python :: 3.5',
'Programming Language :: Python :: 3.6'
]
)
import cron
@cron.interval(minutes=1)
def every_minute():
pass
@cron.scheduled(hour=9)
def at_9am():
pass
# -*- coding: utf-8
from __future__ import unicode_literals, absolute_import
import os
DEBUG = True
USE_TZ = True
BASE_DIR = os.path.dirname(__file__)
# SECURITY WARNING: keep the secret key used in production secret!
SECRET_KEY = 'uoiyd)o$0)pauxb-b%d732h2qwuknn0)#qrhcbx*b^*w6nquwb'
DATABASES = {
'default': {
'ENGINE': 'django.db.backends.sqlite3',
'NAME': ':memory:',
}
}
ROOT_URLCONF = 'tests.urls'
INSTALLED_APPS = [
'django.contrib.auth',
'django.contrib.contenttypes',
'django.contrib.sites',
'cron',
'tests.myapp'
]
TEMPLATES = [
{
'BACKEND': 'django.template.backends.django.DjangoTemplates',
'DIRS': [
os.path.join(BASE_DIR, 'templates')
],
'OPTIONS': {
'context_processors': [
'django.template.context_processors.debug',
'django.template.context_processors.request'
],
'loaders': [
'django.template.loaders.filesystem.Loader',
'django.template.loaders.app_directories.Loader'
],
'debug': DEBUG
}
}
]
SITE_ID = 1
MIDDLEWARE = ()
from django.test import TestCase
from cron import autodiscover, scheduler
class JobTests(TestCase):
def setUp(self):
autodiscover()
def test(self):
jobs = scheduler.get_jobs()
self.assertEqual(len(jobs), 2)
# -*- coding: utf-8 -*-
from __future__ import unicode_literals, absolute_import
urlpatterns = []
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