Skip to content
GitLab
Commit 65a4205d authored by Bram Daams's avatar Bram Daams
Browse files

Resolve "create a pypi package"

parent 445de984
Hide whitespace changes
Inline Side-by-side
.gitignore
View file @ 65a4205d
sch.conf
venv
test.py
__pycache__
SmartCronHelper.egg-info/
# Byte-compiled / optimized / DLL files
__pycache__/
*.py[cod]
*$py.class
# C extensions
*.so
# Distribution / packaging
.Python
env/
build/
develop-eggs/
dist/
downloads/
eggs/
.eggs/
lib/
lib64/
parts/
sdist/
var/
wheels/
*.egg-info/
.installed.cfg
*.egg
# PyInstaller
# Usually these files are written by a python script from a template
# before PyInstaller builds the exe, so as to inject date/other infos into it.
*.manifest
*.spec
# Installer logs
pip-log.txt
pip-delete-this-directory.txt
# Unit test / coverage reports
htmlcov/
.tox/
.coverage
.coverage.*
.cache
nosetests.xml
coverage.xml
*.cover
.hypothesis/
.pytest_cache/
# Translations
*.mo
*.pot
# Django stuff:
*.log
local_settings.py
# Flask stuff:
instance/
.webassets-cache
# Scrapy stuff:
.scrapy
# Sphinx documentation
docs/_build/
# PyBuilder
target/
# Jupyter Notebook
.ipynb_checkpoints
# pyenv
.python-version
# celery beat schedule file
celerybeat-schedule
# SageMath parsed files
*.sage.py
# dotenv
.env
# virtualenv
.venv
venv/
ENV/
# Spyder project settings
.spyderproject
.spyproject
# Rope project settings
.ropeproject
# mkdocs documentation
/site
# mypy
.mypy_cache/
# IDE settings
.vscode/
\ No newline at end of file
.gitlab-ci.yml
View file @ 65a4205d
......@@ -11,10 +11,10 @@ stages:
flake8:
stage: Static Analysis
script:
- flake8 --max-line-length=120 *.py
- flake8 --max-line-length=120 setup.py sch/*.py
pylint:
stage: Static Analysis
allow_failure: true
script:
- pylint -d C0301 *.py
- pylint -d C0301 setup.py sch/*.py
HISTORY.md 0 → 100644
View file @ 65a4205d
## History
#### release notes for v0.1
- initial release
- testing the code on a couple of servers
### release notes for 0.2.0
- changed file structure
- first release on PyPI.
MANIFEST.in 0 → 100644
View file @ 65a4205d
include HISTORY.md
include LICENSE
include README.md
recursive-include tests *
recursive-exclude * __pycache__
recursive-exclude * *.py[co]
recursive-include docs *.md conf.py Makefile make.bat *.jpg *.png *.gif
Makefile 0 → 100644
View file @ 65a4205d
.PHONY: clean clean-test clean-pyc clean-build docs help
.DEFAULT_GOAL := help
define BROWSER_PYSCRIPT
import os, webbrowser, sys
from urllib.request import pathname2url
webbrowser.open("file://" + pathname2url(os.path.abspath(sys.argv[1])))
endef
export BROWSER_PYSCRIPT
define PRINT_HELP_PYSCRIPT
import re, sys
for line in sys.stdin:
match = re.match(r'^([a-zA-Z_-]+):.*?## (.*)$$', line)
if match:
target, help = match.groups()
print("%-20s %s" % (target, help))
endef
export PRINT_HELP_PYSCRIPT
BROWSER := python -c "$$BROWSER_PYSCRIPT"
help:
@python -c "$$PRINT_HELP_PYSCRIPT" < $(MAKEFILE_LIST)
clean: clean-build clean-pyc clean-test ## remove all build, test, coverage and Python artifacts
clean-build: ## remove build artifacts
rm -fr build/
rm -fr dist/
rm -fr .eggs/
find . -name '*.egg-info' -exec rm -fr {} +
find . -name '*.egg' -exec rm -f {} +
clean-pyc: ## remove Python file artifacts
find . -name '*.pyc' -exec rm -f {} +
find . -name '*.pyo' -exec rm -f {} +
find . -name '*~' -exec rm -f {} +
find . -name '__pycache__' -exec rm -fr {} +
clean-test: ## remove test and coverage artifacts
rm -fr .tox/
rm -f .coverage
rm -fr htmlcov/
rm -fr .pytest_cache
lint: ## check style with flake8
flake8 sch tests
pylint sch tests
test: ## run tests quickly with the default Python
python setup.py test
test-all: ## run tests on every Python version with tox
tox
coverage: ## check code coverage quickly with the default Python
coverage run --source sch setup.py test
coverage report -m
coverage html
$(BROWSER) htmlcov/index.html
# docs: ## generate Sphinx HTML documentation, including API docs
# rm -f docs/sch.rst
# rm -f docs/modules.rst
# sphinx-apidoc -o docs/ sch
# $(MAKE) -C docs clean
# $(MAKE) -C docs html
# $(BROWSER) docs/_build/html/index.html
#servedocs: docs ## compile the docs watching for changes
# watchmedo shell-command -p '*.rst' -c '$(MAKE) -C docs html' -R -D .
release: dist ## package and upload a release
twine upload dist/*
dist: clean ## builds source and wheel package
python setup.py sdist
python setup.py bdist_wheel
ls -l dist
install: clean ## install the package to the active Python's site-packages
python setup.py install
README.md
View file @ 65a4205d
......@@ -9,8 +9,7 @@ A cron shell wrapper for registering and updating cron jobs automatically in
Install the `sch` shell and it's dependencies by running pip in the cloned
repository:
``` console
$ cd sch
$ sudo pip install .
$ pip install sch
```
Create a configuration file:
......
requirements_dev.txt 0 → 100644
View file @ 65a4205d
arrow
bump2version==0.5.11
click==7.0
configparser
coverage==4.5.4
flake8==3.7.8
pip==19.2.3
pylint
python-crontab
requests
tox==3.14.0
ttictoc
twine==1.14.0
tzlocal
watchdog==0.9.0
wheel==0.33.6
sch/Cron.py 0 → 100644
View file @ 65a4205d
"""
Wrapper around CronTabs
"""
import logging
from crontabs import CronTabs
from sch import Job
class Cron():
"""
Cron searches for cron jobs with the environment variable
"JOB_ID={job_id}" for given job_id
"""
# pylint: disable=too-few-public-methods
def __init__(self, job_id):
self._jobs = []
self._job_id = job_id
command_filter = "JOB_ID={} ".format(job_id)
crontabs = CronTabs().all.find_command(command_filter)
for crontab in crontabs:
if crontab.enabled:
self._jobs.append(Job(crontab))
def job(self):
"""
returns the matching cron job
or None if there are no or multiple matches or
if given job_id was None to start with
"""
if not self._job_id:
return None
if len(self._jobs) == 1:
return self._jobs[0]
logging.error(
'found %s matching cron jobs for given job id'
'. 1 expected (job.id=%s)',
len(self._jobs),
self._job_id
)
return None
hc.py → sch/Healthchecks.py
View file @ 65a4205d
"""
module for interfacing a healthchecks.io compatible service
"""
import collections
import hashlib
import logging
import json
import os
......@@ -14,197 +12,6 @@ import arrow
import click
import requests
import tzlocal
from crontabs import CronTabs
HealthcheckCredentials = collections.namedtuple(
'HealthcheckCredentials',
'api_url api_key'
)
INTERVAL_DICT = collections.OrderedDict([
("Y", 365*86400), # 1 year
("M", 30*86400), # 1 month
("W", 7*86400), # 1 week
("D", 86400), # 1 day
("h", 3600), # 1 hour
("m", 60), # 1 minute
("s", 1)]) # 1 second
class Cron():
"""
Cron searches for cron jobs with the environment variable
"JOB_ID={job_id}" for given job_id
"""
# pylint: disable=too-few-public-methods
def __init__(self, job_id):
self._jobs = []
self._job_id = job_id
command_filter = "JOB_ID={} ".format(job_id)
crontabs = CronTabs().all.find_command(command_filter)
for crontab in crontabs:
if crontab.enabled:
self._jobs.append(Job(crontab))
def job(self):
"""
returns the matching cron job
or None if there are no or multiple matches or
if given job_id was None to start with
"""
if not self._job_id:
return None
if len(self._jobs) == 1:
return self._jobs[0]
logging.error(
'found %s matching cron jobs for given job id'
'. 1 expected (job.id=%s)',
len(self._jobs),
self._job_id
)
return None
class Job():
"""
Wrapper to create a self aware cron job object
"""
# pylint does not like the number of attributes and
# public methods, but i do ;-)
# pylint: disable=too-many-instance-attributes
# pylint: disable=too-few-public-methods
def __init__(self, job):
# wrab the job
self._job = job
self.id = self._get_id() # pylint: disable=invalid-name
self.command = self._job.command
self.comment = self._job.comment
self.tags = self._get_tags()
self.schedule = self._get_schedule()
self.grace = self._get_grace()
# finally, determine hash
self.hash = self._get_hash()
def _get_env_var(self, env_var):
"""
Returns the value of an environment variable
"""
regex = r".*{env_var}=([\w,-]*)".format(env_var=env_var)
match = re.match(regex, self._job.command)
if match:
return match.group(1)
return None
def _get_id(self):
"""
Returns the value of environment variable JOB_ID if specified
in the cron job
"""
return self._get_env_var('JOB_ID')
def _get_tags(self):
"""
Returns the tags specified in the environment variable
JOB_TAGS in the cron job
"""
tags = self._get_env_var('JOB_TAGS')
if tags:
return tags.replace(',', ' ')
return ""
def _get_schedule(self):
"""
extract the schedule in 5 column notation from the given job
"""
# correct schedule aliases back to fields
schedule = self._job.slices.render()
if schedule == '@hourly':
schedule = '0 * * * *'
if schedule == '@daily':
schedule = '0 0 * * *'
if schedule == '@weekly':
schedule = '0 0 * * 0'
if schedule == '@monthly':
schedule = '0 0 1 * *'
if schedule == '@yearly':
schedule = '0 0 1 1 *'
return schedule
def _get_hash(self):
"""Returns the unique hash for given cron job"""
md5 = hashlib.md5()
# job schedule
md5.update(self.schedule.encode('utf-8'))
# the command itself
md5.update(self.command.encode('utf-8'))
# the comment
md5.update(self.comment.encode('utf-8'))
# host fqdn
md5.update(socket.getfqdn().encode('utf-8'))
# job user
md5.update(os.environ['LOGNAME'].encode('utf-8'))
# the timezone (not so likely to change)
md5.update(tzlocal.get_localzone().zone.encode('utf-8'))
return md5.hexdigest()
def _get_grace(self):
"""
Returns the jobs grace time in seconds as specified by the
commands' environment variable JOB_GRACE
"""
grace = self._get_env_var('JOB_GRACE')
if grace:
grace = self._human_to_seconds(grace)
return grace
return None
@staticmethod
def _human_to_seconds(string):
"""Convert internal string like 1M, 1Y3M, 3W to seconds.
:type string: str
:param string: Interval string like 1M, 1W, 1M3W4h2s...
(s => seconds, m => minutes, h => hours, D => days,
W => weeks, M => months, Y => Years).
:rtype: int
:return: The conversion in seconds of string.
"""
interval_exc = "Bad interval format for {0}".format(string)
interval_regex = re.compile(
"^(?P<value>[0-9]+)(?P<unit>[{0}])".format(
"".join(INTERVAL_DICT.keys())))
if string.isdigit():
seconds = int(string)
return seconds
seconds = 0
while string:
match = interval_regex.match(string)
if match:
value, unit = int(match.group("value")), match.group("unit")
if int(value) and unit in INTERVAL_DICT:
seconds += value * INTERVAL_DICT[unit]
string = string[match.end():]
else:
raise Exception(interval_exc)
else:
raise Exception(interval_exc)
return seconds
class Healthchecks:
......
sch/HealthchecksCredentials.py 0 → 100644
View file @ 65a4205d
"""
"""
import collections
HealthchecksCredentials = collections.namedtuple(
'HealthchecksCredentials',
'api_url api_key'
)
sch/Job.py 0 → 100644
View file @ 65a4205d
"""
module for interfacing a healthchecks.io compatible service
"""
import collections
import hashlib
import os
import re
import socket
import tzlocal
INTERVAL_DICT = collections.OrderedDict([
("Y", 365*86400), # 1 year
("M", 30*86400), # 1 month
("W", 7*86400), # 1 week
("D", 86400), # 1 day
("h", 3600), # 1 hour
("m", 60), # 1 minute
("s", 1)]) # 1 second
class Job():
"""
Wrapper to create a self aware cron job object
"""
# pylint does not like the number of attributes and
# public methods, but i do ;-)
# pylint: disable=too-many-instance-attributes
# pylint: disable=too-few-public-methods
def __init__(self, job):
# wrab the job
self._job = job
self.id = self._get_id() # pylint: disable=invalid-name
self.command = self._job.command
self.comment = self._job.comment
self.tags = self._get_tags()
self.schedule = self._get_schedule()
self.grace = self._get_grace()
# finally, determine hash
self.hash = self._get_hash()
def _get_env_var(self, env_var):
"""
Returns the value of an environment variable
"""
regex = r".*{env_var}=([\w,-]*)".format(env_var=env_var)
match = re.match(regex, self._job.command)
if match:
return match.group(1)
return None
def _get_id(self):
"""
Returns the value of environment variable JOB_ID if specified
in the cron job
"""
return self._get_env_var('JOB_ID')
def _get_tags(self):
"""
Returns the tags specified in the environment variable
JOB_TAGS in the cron job
"""
tags = self._get_env_var('JOB_TAGS')
if tags:
return tags.replace(',', ' ')
return ""
def _get_schedule(self):
"""
extract the schedule in 5 column notation from the given job
"""
# correct schedule aliases back to fields
schedule = self._job.slices.render()
if schedule == '@hourly':
schedule = '0 * * * *'
if schedule == '@daily':
schedule = '0 0 * * *'
if schedule == '@weekly':
schedule = '0 0 * * 0'
if schedule == '@monthly':
schedule = '0 0 1 * *'
if schedule == '@yearly':
schedule = '0 0 1 1 *'
return schedule
def _get_hash(self):
"""Returns the unique hash for given cron job"""
md5 = hashlib.md5()
# job schedule
md5.update(self.schedule.encode('utf-8'))
# the command itself
md5.update(self.command.encode('utf-8'))
# the comment
md5.update(self.comment.encode('utf-8'))
# host fqdn
md5.update(socket.getfqdn().encode('utf-8'))
# job user
md5.update(os.environ['LOGNAME'].encode('utf-8'))
# the timezone (not so likely to change)
md5.update(tzlocal.get_localzone().zone.encode('utf-8'))
return md5.hexdigest()
def _get_grace(self):
"""
Returns the jobs grace time in seconds as specified by the
commands' environment variable JOB_GRACE
"""
grace = self._get_env_var('JOB_GRACE')
if grace:
grace = self._human_to_seconds(grace)
return grace