<s> """<STR_LIT>""" <EOL> import json <EOL> import requests <EOL> from datetime import datetime <EOL> from django . conf import settings <EOL> from httpsig . requests_auth import HTTPSignatureAuth <EOL> from time import mktime <EOL> from wsgiref . handlers import format_date_time <EOL> import logging <EOL> logger = logging . getLogger ( __name__ ) <EOL> _httpsig_auth = HTTPSignatureAuth ( key_id = settings . ORCHESTRA_PROJECT_API_KEY , <EOL> secret = settings . ORCHESTRA_PROJECT_API_SECRET , <EOL> algorithm = '<STR_LIT>' ) <EOL> _api_root_url = '<STR_LIT>' . format ( settings . ORCHESTRA_URL ) <EOL> class OrchestraError ( Exception ) : <EOL> pass <EOL> def _make_api_request ( method , endpoint , * args , ** kwargs ) : <EOL> func = getattr ( requests , method ) <EOL> headers = { '<STR_LIT:date>' : format_date_time ( mktime ( datetime . now ( ) . timetuple ( ) ) ) , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : settings . ORCHESTRA_PROJECT_API_KEY } <EOL> headers . update ( kwargs . pop ( '<STR_LIT>' , { } ) ) <EOL> all_kwargs = { '<STR_LIT>' : _httpsig_auth , '<STR_LIT>' : headers } <EOL> all_kwargs . update ( kwargs ) <EOL> url = '<STR_LIT>' . format ( _api_root_url , endpoint ) <EOL> response = func ( url , * args , ** all_kwargs ) <EOL> if response . status_code != <NUM_LIT:200> : <EOL> raise OrchestraError ( response . text ) <EOL> return response <EOL> def get_workflow_types ( ) : <EOL> response = _make_api_request ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> return json . loads ( response . text ) [ '<STR_LIT>' ] <EOL> def get_project_details_url ( project_id ) : <EOL> data = { '<STR_LIT>' : project_id } <EOL> response = _make_api_request ( '<STR_LIT>' , '<STR_LIT>' , <EOL> data = json . dumps ( data ) ) <EOL> url = json . loads ( response . text ) [ '<STR_LIT>' ] <EOL> return url <EOL> def create_orchestra_project ( client , <EOL> workflow_slug , <EOL> workflow_version_slug , <EOL> description , <EOL> priority , <EOL> project_data , <EOL> task_class ) : <EOL> data = { <EOL> '<STR_LIT>' : workflow_slug , <EOL> '<STR_LIT>' : workflow_version_slug , <EOL> '<STR_LIT:description>' : description , <EOL> '<STR_LIT>' : priority , <EOL> '<STR_LIT>' : project_data , <EOL> '<STR_LIT>' : task_class <EOL> } <EOL> response = _make_api_request ( '<STR_LIT>' , '<STR_LIT>' , <EOL> data = json . dumps ( data ) ) <EOL> project_id = json . loads ( response . text ) [ '<STR_LIT>' ] <EOL> return project_id <EOL> def get_project_information ( project_id ) : <EOL> data = { <EOL> '<STR_LIT>' : project_id <EOL> } <EOL> response = _make_api_request ( '<STR_LIT>' , '<STR_LIT>' , <EOL> data = json . dumps ( data ) ) <EOL> return json . loads ( response . text ) </s>
