tyrel
/
celigo
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

setup basic project, with code in core. setup.py, make file, start of a readme.

This commit is contained in:
Tyrel Souza 2016-05-19 10:43:19 -04:00
parent 78f6ac9347
commit 5cccc4f8e2
No known key found for this signature in database
GPG Key ID: 2EECB5087209E6A5
8 changed files with 356 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
Makefile Normal file
View File

@ -0,0 +1,5 @@
init:
pip install -r requirements.txt
test:
py.test tests

7
README.rst Normal file
View File

@ -0,0 +1,7 @@
Celigo Backup
-------------
This module interacts with the Celigo API.
If you have any Celigo flows that have a NetSuite connection, and you wish
to backup the field mappings for the flow, then use this.

1
celigo/__init__.py Normal file
View File

@ -0,0 +1 @@
from .core import BackupCeligo

211
celigo/core.py Normal file
View File

@ -0,0 +1,211 @@
import glob
import logging
import requests
import io
import os
import json
from slugify import slugify
import prompt
L = logging.getLogger(__name__)
class BackupCeligo(object):
"""
This module interacts with the Celigo API.
"""
def __init__(self, data_dir, api_key, base_url=None):
if not base_url:
base_url = "https://api.integrator.io/v1/"
self.api_key = api_key
self.data_dir = data_dir
self.base_url = base_url
self.imports_cache = {}
self.setup_requests_session()
self.ensure_directories_exist()
def ensure_directories_exist(self):
""" Make the directory if it doesn't exist """
subdirs = ('imports', 'connections')
for subdir in subdirs:
_dir = os.path.join(self.data_dir, subdir)
if not os.path.exists(_dir):
os.makedirs(_dir)
def setup_requests_session(self):
self.session = requests.Session()
self.session.headers.update({
"Authorization": "Bearer {API_KEY}".format(
API_KEY=self.api_key),
"Content-Type": "application/json"
}
)
def _celigo_api_get(self, path):
"""
Make a GET request to the celigo API
:param path: The rest of the path after BASE_URL
:return conf_dict: response as a dict of the json returned from the
api.
"""
response = self.session.get(url=self.base_url + path)
response.raise_for_status()
conf_dict = response.json()
L.info("Got conf_dict from %s", self.base_url + path)
return conf_dict
def _celigo_api_put(self, path, body):
"""
Make a GET request to the celigo API
:param path: The rest of the path after BASE_URL
:return conf_dict: response as a dict of the json returned from the
api.
"""
response = self.session.put(
url=self.base_url + path,
data=json.dumps(body)
)
response.raise_for_status()
L.info("Restored backup to %s", self.base_url + path)
return response
def backup(self, auto=False):
"""
Get all the flow data from Celigo.
Then loop over each flow and cache it's Import data in an instance
varable.
Once this is cached, save the imports.
"""
try:
flows = self._celigo_api_get("flows/")
for flow in flows:
self.cache_import_remote(flow)
except requests.exceptions.RequestException:
L.info('HTTP Request failed')
raise
L.info("Got all imports, writing now")
L.info("We have imports for: %s", ", ".join(self.imports_cache.keys()))
for flow_name, (import_id, import_conf) in self.imports_cache.items():
self.save_import(flow_name, auto)
def restore(self, auto=False):
"""
Get all the import files in the import direcotry,
and store them in the local self.imports_cache cache.
Once they are stored, prompt on which to restore to Celigo.
"""
import_dir = os.path.join(self.data_dir, "imports")
for fname in glob.glob("{}/*_*.json".format(import_dir)):
self.cache_import_json(fname)
while self.imports_cache.keys():
if auto:
action = "All"
else:
options = self.imports_cache.keys()
options = sorted(options)
options.append("All")
options.append("None")
action = prompt("Backup which import(s)?", options)
if action == "None":
# We're done here.
return
elif action == "All":
# Process them all.
for key in self.imports_cache.keys():
self.restore_to_celigo(key)
break
else:
# Process one and remove it from the cache so we don't reupload
self.restore_to_celigo(action)
del self.imports_cache[action]
def cache_import_remote(self, flow):
"""
Stores the import in self.imports_cache before write.
"""
flow_name = slugify(flow['name'])
import_id = flow['_importId']
import_conf = self._celigo_api_get(
"imports/{id}/distributed".format(
id=import_id))
self.imports_cache[flow_name] = (import_id, import_conf)
def save_import(self, flow_name, auto=False):
"""
Write the import to a .json file with name_id.json format.
Prompt for overwrite.
:param flow_name: the slugified name of the flow as a key
:param auto: if auto is true, don't prompt for overwrite
"""
import_id, import_conf = self.imports_cache[flow_name]
filename = os.path.join(
self.data_dir,
"imports",
"%s_%s.json" % (flow_name, import_id))
write = True
# By default, we prompt for overwrites
if os.path.isfile(filename) and not auto:
# Check that we should overwrite
overwrite = prompt(
"File {filename} exists, overwrite file?".format(
filename=filename)
)
write = bool(overwrite == "Yes")
if write:
self.write_json(filename, import_conf)
else:
L.info("You chose not to save this file.")
def write_json(self, filename, contents):
"""
Write a file of json data and say that we wrote it.
"""
with io.open(filename, "w", encoding='utf-8') as f:
f.write(unicode(json.dumps(contents, f, indent=4,
ensure_ascii=False, encoding='utf-8')))
L.info("Wrote {}".format(filename))
def cache_import_json(self, fname):
"""
Load the import from a .json file with name_id.json format.
"""
with io.open(fname, "r", encoding='utf-8') as f:
json_data = json.load(f)
fname = os.path.split(fname)[1][:-5]
flow_name, import_id = fname.split("_")
# Store locally in the instance cache
self.imports_cache[flow_name] = (import_id, json_data)
def restore_to_celigo(self, key):
"""
Make the call to celigo to update the data
:params key: slugified name to use as a key
"""
raise NotImplemented("NOT IMPLEMENTED YET")
# yes this works but I don't want to allow this in testing.
import_id, import_conf = self.imports_cache[key]
response = self._celigo_api_put(
"imports/{id}/distributed".format(
id=import_id),
import_conf)
L.info("Restored: ")
L.info(response.json()['_id'])
L.info("")

41
celigo/prompt.py Normal file
View File

@ -0,0 +1,41 @@
import collections
def _get_input():
""" Putting in a method so we can mock it easier"""
return raw_input("(Type a number) (Ctrl+c to cancel): ").rstrip("\n")
def prompt(message, options=None, retries=3):
"""
Accept a list of options (or defaults to Yes/No) and continue asking
until one of the appropriate answers are accepted, or failing after
3 times.
:params message: The message to show as a prompt
:params options: An iterator of strings to show as each question in the
prompt
:return selected: Return the selected answer from options.
"""
if options is None:
options = ("Yes", "No")
mapping = collections.OrderedDict()
for idx, v in enumerate(options):
mapping[str(idx + 1)] = v
option_strings = "\n".join(" {0} - {1}".format(k, v)
for k, v in mapping.items())
print(message)
print(option_strings)
option = None
while True:
try:
index = _get_input()
option = mapping[index]
except KeyError:
print("Invalid Option, try again (Ctrl+C to cancel)")
retries -= 1
if retries:
continue
else:
break
break
return option

0
README.md → celigo/tests.py
View File

15
requirements.txt Normal file
View File

@ -0,0 +1,15 @@
flake8==2.5.4
mccabe==0.4.0
pbr==1.9.1
pep8==1.7.0
py==1.4.31
pyflakes==1.0.0
pytest==2.9.1
python-slugify==1.2.0
requests==2.10.0
six==1.10.0
stevedore==1.12.0
Unidecode==0.4.19
virtualenv==15.0.1
virtualenv-clone==0.2.6
virtualenvwrapper==4.7.1

76
setup.py Normal file
View File

@ -0,0 +1,76 @@
"""A setuptools based setup module.
See:
https://packaging.python.org/en/latest/distributing.html
https://github.com/pypa/sampleproject
"""
# Always prefer setuptools over distutils
from setuptools import setup, find_packages
# To use a consistent encoding
from codecs import open
from os import path
here = path.abspath(path.dirname(__file__))
# Get the long description from the README file
with open(path.join(here, 'README.rst'), encoding='utf-8') as f:
long_description = f.read()
setup(
name='celigo',
# Versions should comply with PEP440. For a discussion on single-sourcing
# the version across setup.py and the project code, see
# https://packaging.python.org/en/latest/single_source_version.html
version='1.2.0',
description='Celigo API interaction',
long_description=long_description,
# The project's main homepage.
url='https://gitlab.com/tyrelsouza/celigo',
# Author details
author='Tyrel Souza',
author_email='tyrel@tyrelsouza.com',
# See https://pypi.python.org/pypi?%3Aaction=list_classifiers
classifiers=[
# How mature is this project? Common values are
# 3 - Alpha
# 4 - Beta
# 5 - Production/Stable
'Development Status :: 3 - Alpha',
# Indicate who your project is intended for
'Intended Audience :: Developers',
# Specify the Python versions you support here. In particular, ensure
# that you indicate whether you support Python 2, Python 3 or both.
'Programming Language :: Python :: 2',
'Programming Language :: Python :: 2.7',
'Programming Language :: Python :: 3',
'Programming Language :: Python :: 3.3',
'Programming Language :: Python :: 3.4',
'Programming Language :: Python :: 3.5',
],
# What does your project relate to?
keywords='api celigo netsuite',
# You can just specify the packages manually here if your project is
# simple. Or you can use find_packages().
packages=['celigo'],
# Alternatively, if you want to distribute just a my_module.py, uncomment
# this:
# py_modules=["my_module"],
# List run-time dependencies here. These will be installed by pip when
# your project is installed. For an analysis of "install_requires" vs pip's
# requirements files see:
# https://packaging.python.org/en/latest/requirements.html
install_requires=['requirements',
'python-slugify'],
)