<commit_before>"""hug/use.py

Provides a mechanism for using external hug APIs both locally or remotely in a seamless fashion

Copyright (C) 2015  <NAME>

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.

"""
from collections import namedtuple

from io import BytesIO
import requests

import hug._empty as empty
from hug.defaults import input_format
from hug.input_format import separate_encoding

Response = namedtuple('Response', ['content', 'status_code', 'headers'])


class Service(object):
    """Defines the base concept of a consumed service.
        This is to enable encapsulating the logic of calling a service so usage can be independant of the interface
    """
    __slots__ = ('timeout', 'raise_on', 'version')

    def __init__(self, version=None, timeout=None, raise_on=(500, )):
        self.version = version
        self.timeout = timeout
        self.raise_on = raise_on if type(raise_on) in (tuple, list) else (raise_on, )

    def request(self, method, url, headers=empty.dict, timeout=None, **params):
        """Calls the service at the specified URL using the "CALL" method"""
        raise NotImplementedError("Concrete services must define the request method")

    def get(self, url, headers=empty.dict, timeout=None, **params):
        """Calls the service at the specified URL using the "GET" method"""
        return self.request('GET', url=url, headers=headers, timeout=timeout, **params)

    def post(self, url, headers=empty.dict, timeout=None, **params):
        """Calls the service at the specified URL using the "POST" method"""
        return self.request('POST', url=url, headers=headers, timeout=timeout, **params)

    def delete(self, url, headers=empty.dict, timeout=None, **params):
        """Calls the service at the specified URL using the "DELETE" method"""
        return self.request('DELETE', url=url, headers=headers, timeout=timeout, **params)

    def put(self, url, headers=empty.dict, timeout=None, **params):
        """Calls the service at the specified URL using the "PUT" method"""
        return self.request('PUT', url=url, headers=headers, timeout=timeout, **params)

    def trace(self, url, headers=empty.dict, timeout=None, **params):
        """Calls the service at the specified URL using the "TRACE" method"""
        return self.request('TRACE', url=url, headers=headers, timeout=timeout, **params)

    def patch(self, url, headers=empty.dict, timeout=None, **params):
        """Calls the service at the specified URL using the "PATCH" method"""
        return self.request('PATCH', url=url, headers=headers, timeout=timeout, **params)

    def options(self, url, headers=empty.dict, timeout=None, **params):
        """Calls the service at the specified URL using the "OPTIONS" method"""
        return self.request('OPTIONS', url=url, headers=headers, timeout=timeout, **params)

    def head(self, url, headers=empty.dict, timeout=None, **params):
        """Calls the service at the specified URL using the "HEAD" method"""
        return self.request('HEAD', url=url, headers=headers, timeout=timeout, **params)

    def connect(self, url, headers=empty.dict, timeout=None, **params):
        """Calls the service at the specified URL using the "CONNECT" method"""
        return self.request('CONNECT', url=url, headers=headers, timeout=timeout, **params)


class HTTP(Service):
    __slots__ = ('endpoint', 'session')

    def __init__(self, endpoint, auth=None, headers=empty.dict, version=None, timeout=None, raise_on=(500, )):
        super().__init__(timeout=timeout, raise_on=raise_on, version=version)
        self.endpoint = endpoint
        self.session = requests.Session()
        self.session.auth = auth
        self.session.headers.update(headers)

    def request(self, method, url, headers=empty.dict, timeout=None, **params):
        if self.version:
            url = "/{0}/{1}".format(self.version, url)
        response = self.session.request(method, self.endpoint + url, headers=headers, params=params)

        data = BytesIO(response.content)
        (content_type, encoding) = separate_encoding(response.headers.get('Content-Type', ''), 'utf-8')
        if content_type in input_format:
            data = input_format[content_type](data, encoding)

        if response.status_code in self.raise_on:
            raise requests.HTTPError('{0} {1} occured for url: {2}'.format(response.status_code, response.reason, url))


        return Response(data, response.status_code, response.headers)


class Local(Service):
    __slots__ = ('api', )

    def __init__(self, endpoint, version=None, headers=empty.dict, timeout=None, raise_on=(500, )):

<commit_msg>status<commit_after>"""hug/use.py

Provides a mechanism for using external hug APIs both locally or remotely in a seamless fashion

Copyright (C) 2015  <NAME>

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.

"""
from collections import namedtuple

from io import BytesIO
import requests

import hug._empty as empty
from hug.defaults import input_format
from hug.input_format import separate_encoding

Response = namedtuple('Response', ['content', 'status_code', 'headers'])


class Service(object):
    """Defines the base concept of a consumed service.
        This is to enable encapsulating the logic of calling a service so usage can be independant of the interface
    """
    __slots__ = ('timeout', 'raise_on', 'version')

    def __init__(self, version=None, timeout=None, raise_on=(500, )):
        self.version = version
        self.timeout = timeout
        self.raise_on = raise_on if type(raise_on) in (tuple, list) else (raise_on, )

    def request(self, method, url, headers=empty.dict, timeout=None, **params):
        """Calls the service at the specified URL using the "CALL" method"""
        raise NotImplementedError("Concrete services must define the request method")

    def get(self, url, headers=empty.dict, timeout=None, **params):
        """Calls the service at the specified URL using the "GET" method"""
        return self.request('GET', url=url, headers=headers, timeout=timeout, **params)

    def post(self, url, headers=empty.dict, timeout=None, **params):
        """Calls the service at the specified URL using the "POST" method"""
        return self.request('POST', url=url, headers=headers, timeout=timeout, **params)

    def delete(self, url, headers=empty.dict, timeout=None, **params):
        """Calls the service at the specified URL using the "DELETE" method"""
        return self.request('DELETE', url=url, headers=headers, timeout=timeout, **params)

    def put(self, url, headers=empty.dict, timeout=None, **params):
        """Calls the service at the specified URL using the "PUT" method"""
        return self.request('PUT', url=url, headers=headers, timeout=timeout, **params)

    def trace(self, url, headers=empty.dict, timeout=None, **params):
        """Calls the service at the specified URL using the "TRACE" method"""
        return self.request('TRACE', url=url, headers=headers, timeout=timeout, **params)

    def patch(self, url, headers=empty.dict, timeout=None, **params):
        """Calls the service at the specified URL using the "PATCH" method"""
        return self.request('PATCH', url=url, headers=headers, timeout=timeout, **params)

    def options(self, url, headers=empty.dict, timeout=None, **params):
        """Calls the service at the specified URL using the "OPTIONS" method"""
        return self.request('OPTIONS', url=url, headers=headers, timeout=timeout, **params)

    def head(self, url, headers=empty.dict, timeout=None, **params):
        """Calls the service at the specified URL using the "HEAD" method"""
        return self.request('HEAD', url=url, headers=headers, timeout=timeout, **params)

    def connect(self, url, headers=empty.dict, timeout=None, **params):
        """Calls the service at the specified URL using the "CONNECT" method"""
        return self.request('CONNECT', url=url, headers=headers, timeout=timeout, **params)


class HTTP(Service):
    __slots__ = ('endpoint', 'session')

    def __init__(self, endpoint, auth=None, version=None, headers=empty.dict, timeout=None, raise_on=(500, )):
        super().__init__(timeout=timeout, raise_on=raise_on, version=version)
        self.endpoint = endpoint
        self.session = requests.Session()
        self.session.auth = auth
        self.session.headers.update(headers)

    def request(self, method, url, headers=empty.dict, timeout=None, **params):
        if self.version:
            url = "/{0}/{1}".format(self.version, url)
        response = self.session.request(method, self.endpoint + url, headers=headers, params=params)

        data = BytesIO(response.content)
        (content_type, encoding) = separate_encoding(response.headers.get('Content-Type', ''), 'utf-8')
        if content_type in input_format:
            data = input_format[content_type](data, encoding)

        if response.status_code in self.raise_on:
            raise requests.HTTPError('{0} {1} occured for url: {2}'.format(response.status_code, response.reason, url))


        return Response(data, response.status_code, response.headers)


class Local(Service):
    __slots__ = ('api', 'headers')

    def __init__(self, api, version=None, headers=empty.dict, timeout=None, raise_on=(500, )):
        super().__init__(timeout=timeout, raise_on=raise_on, version=version)
        self.api = api
        self.headers = headers

{'fraction_non_alphanumeric': 0.11089664853483067, 'fraction_numerical': 0.0033339182312686435, 'mean_word_length': 4.097942754919499, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7500831', 'n_tokens_mistral': 3121, 'n_tokens_neox': 3018, 'n_words': 1160}