Commit 2d18625f authored by Christian Lölkes's avatar Christian Lölkes
Browse files

Version 1.0.0

parents
# Byte-compiled / optimized / DLL files
__pycache__/
*.py[cod]
*$py.class
# C extensions
*.so
# Distribution / packaging
.Python
build/
develop-eggs/
dist/
downloads/
eggs/
.eggs/
lib/
lib64/
parts/
sdist/
var/
wheels/
pip-wheel-metadata/
share/python-wheels/
*.egg-info/
.installed.cfg
*.egg
MANIFEST
# 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/
.nox/
.coverage
.coverage.*
.cache
nosetests.xml
coverage.xml
*.cover
*.py,cover
.hypothesis/
.pytest_cache/
# Translations
*.mo
*.pot
# Django stuff:
*.log
local_settings.py
db.sqlite3
db.sqlite3-journal
# Flask stuff:
instance/
.webassets-cache
# Scrapy stuff:
.scrapy
# Sphinx documentation
docs/_build/
# PyBuilder
target/
# Jupyter Notebook
.ipynb_checkpoints
# IPython
profile_default/
ipython_config.py
# pyenv
.python-version
# pipenv
# According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
# However, in case of collaboration, if having platform-specific dependencies or dependencies
# having no cross-platform support, pipenv may install dependencies that don't work, or not
# install all needed dependencies.
#Pipfile.lock
# celery beat schedule file
celerybeat-schedule
# SageMath parsed files
*.sage.py
# Environments
.env
.venv
env/
venv/
ENV/
env.bak/
venv.bak/
# Spyder project settings
.spyderproject
.spyproject
# Rope project settings
.ropeproject
# mkdocs documentation
/site
# mypy
.mypy_cache/
.dmypy.json
dmypy.json
# Pyre type checker
.pyre/
# Changelog
All notable changes to this project will be documented in this file.
The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
## [Unreleased]
## [1.0.0] 2019-10-7
- Initial release
Copyright 2017 Christian Lölkes
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.
# obelixtools
1. Install it with ```pip install obelixtools```
## API
```from obelixtools import API```
This is a small wrapper I wrote around the reuqests library. It has some extra features like a speedtestand connectivity check.
import time
import json
import requests
from sseclient import SSEClient
import xml.etree.ElementTree as ET
import logging
logger = logging.getLogger(__name__)
class API(object):
def __init__(self, url=None, format=None, user=None, key=None):
self.setupRequests(url=url, user=user, key=key)
self.format = format
def setupRequests(self, url, user=None, key=None, rateLimit=3600, api_suffix=''):
"""
Setup the request parameters.
This should be called once within __init__
Parametes
---------
url: string
The URL for the requests library.
user : string, optional
username for HTTPBasicAuth
key : string, optional
key or password for HTTPBasicAuth
rateLimit : integer, optional
How many seconds to wait betrween two requests. 3600s default.
api_suffix : string, optional
String to append to the URL for the request. Example: '&page=1'
"""
self.api_url = url
self.rateLimit = rateLimit
self.lastUpdate = 0
self.api_suffix = ''
self.request_url = ''
self.status = False
self.auth = requests.auth.HTTPBasicAuth(user, key) if user and key else None
def preQuery(self):
"""
This is called before the requests methods.
The self.query() will use the self.request_url. It can be modified here.
"""
self.request_url = self.api_url
pass
def postQuery(self):
"""
This is called after a successfull reuqest and before the end of the
query() methodself.
The data from the requests lies in self.response. This method can be
used to check the data for validity. self.query() sets self.status to
True before callig this and returns it after this. If the data check
fails set self.status to false.
"""
pass
def stream(self):
try:
response = requests.get(self.request_url, auth=self.auth, stream=True)
except Excpetion as e:
logging.warning(e)
pass
if response.status_code == 200:
self.content = None
self.sseclient = SSEClient(response)
def query(self, url=False):
"""
Perfom a GET-Request on the URL in self.request_url.
Returns
-------
bool
True if successfull, False otherwise.
True means that new data was fetched from the URL. False means the
request has failed or it hits the rate limit.
"""
if time.time() - self.lastUpdate > self.rateLimit or url:
# Ignore rate limit if manual url
self.status = False
self.preQuery()
try:
response = requests.get(url or self.request_url, auth=self.auth)
except Exception as e:
logger.warning(e)
return self.status
self.lastUpdate += 5 if response.status_code == 202 else 0
if response.status_code == 200:
self.handleResponse(response)
self.postQuery()
else:
pass
else:
self.postQuery()
return self.status
def handleResponse(self, response):
if self.format == 'json':
self.content = response.json()
elif self.format == 'xml':
self.content = ET.fromstring(response.content)
else:
self.content = response.content
self.lastUpdate = time.time()
self.status = True
def update(self):
pass
def checkConnection(self, url='https://1.1.1.1', timeout=5):
logger.info('Performing selftest with {}'.format(url))
if self.query(url):
logger.info('Connected to the internet.')
return True
else:
logger.warning('Not connected to the internet.')
return False
def speedtest(self, url='http://speedtest.belwue.net/100M'):
logger.info('Performing speedtest with {}'.format(url))
now = time.time()
if self.query(url):
self.connectionSpeed = int(len(self.content) / (time.time() - now))
logger.info('Connection speed is {}/s'.format(human_readable(self.connectionSpeed)))
return self.connectionSpeed
else:
return False
# Source https://stackoverflow.com/a/43750422
def human_readable(bytes, units=[' bytes','KB','MB','GB','TB', 'PB', 'EB']):
""" Returns a human readable string reprentation of bytes"""
return str(bytes) + units[0] if bytes < 1024 else human_readable(bytes>>10, units[1:])
if __name__ == '__main__':
logging.basicConfig(level=logging.INFO)
logger.info('Script is executed as standalone file.')
test = API()
test.checkConnection()
test.speedtest()
import setuptools
with open('README.md', 'r') as fh:
long_description = fh.read()
setuptools.setup(
name='obelixtools',
version='1.0.0',
author='Christian Lölkes',
author_email='christian.loelkes@gmail.com',
description='Useful tools I need in many projects.',
long_description=long_description,
long_description_content_type='text/markdown',
url='https://github.com/loelkes/pypi-obelixtools',
packages=setuptools.find_packages(),
classifiers=[
'Programming Language :: Python :: 3',
'License :: OSI Approved :: MIT License',
'Operating System :: OS Independent',
],
python_requires='>=3.6',
install_requires=[
'requests',
'sseclient'
],
)
Supports Markdown
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