Подтвердить что ты не робот

Запросы Python извлекают файл из локального URL-адреса

Я использую Python requests в одном из методов моего приложения. Тело метода выглядит следующим образом:

def handle_remote_file(url, **kwargs):
    response = requests.get(url, ...)
    buff = StringIO.StringIO()
    buff.write(response.content)
    ...
    return True

Я хотел бы написать некоторые модульные тесты для этого метода, однако, что я хочу сделать, это передать поддельный локальный url, например:

class RemoteTest(TestCase):
    def setUp(self):
        self.url = 'file:///tmp/dummy.txt'

    def test_handle_remote_file(self):
        self.assertTrue(handle_remote_file(self.url))

Когда я вызываю request.get с локальным url, я получил следующее KeyError:

requests.get('file:///tmp/dummy.txt')

/Library/Frameworks/Python.framework/Versions/2.7/lib/python2.7/site-packages/requests/packages/urllib3/poolmanager.pyc in connection_from_host(self, host, port, scheme)
76 
77         # Make a fresh ConnectionPool of the desired type
78         pool_cls = pool_classes_by_scheme[scheme]
79         pool = pool_cls(host, port, **self.connection_pool_kw)
80 

KeyError: 'file'

Вопрос в том, как передать локальный url в request.get?

PS: Я составил вышеприведенный пример. Возможно, он содержит много ошибок.

4b9b3361

Ответ 1

Поскольку @WooParadog объясняет, что библиотека запросов не знает, как обрабатывать локальные файлы. Хотя текущая версия позволяет определить транспортные адаптеры.

Поэтому вы можете просто определить собственный адаптер, который сможет обрабатывать локальные файлы, например:

from requests_testadapter import Resp

class LocalFileAdapter(requests.adapters.HTTPAdapter):
    def build_response_from_file(self, request):
        file_path = request.url[7:]
        with open(file_path, 'rb') as file:
            buff = bytearray(os.path.getsize(file_path))
            file.readinto(buff)
            resp = Resp(buff)
            r = self.build_response(request, resp)

            return r

    def send(self, request, stream=False, timeout=None,
             verify=True, cert=None, proxies=None):

        return self.build_response_from_file(request)

requests_session = requests.session()
requests_session.mount('file://', LocalFileAdapter())
requests_session.get('file://<some_local_path>')

Я использую requests-testadapter модуль в приведенном выше примере.

Ответ 2

Вот транспортный адаптер, который я написал более функциональный, чем b1r3k, и не имеет дополнительных зависимостей за пределами самих запросов. Я еще не тестировал его исчерпывающе, но то, что я пробовал, кажется, не содержит ошибок.

import requests
import os, sys

if sys.version_info.major < 3:
    from urllib import url2pathname
else:
    from urllib.request import url2pathname

class LocalFileAdapter(requests.adapters.BaseAdapter):
    """Protocol Adapter to allow Requests to GET file:// URLs

    @todo: Properly handle non-empty hostname portions.
    """

    @staticmethod
    def _chkpath(method, path):
        """Return an HTTP status for the given filesystem path."""
        if method.lower() in ('put', 'delete'):
            return 501, "Not Implemented"  # TODO
        elif method.lower() not in ('get', 'head'):
            return 405, "Method Not Allowed"
        elif os.path.isdir(path):
            return 400, "Path Not A File"
        elif not os.path.isfile(path):
            return 404, "File Not Found"
        elif not os.access(path, os.R_OK):
            return 403, "Access Denied"
        else:
            return 200, "OK"

    def send(self, req, **kwargs):  # pylint: disable=unused-argument
        """Return the file specified by the given request

        @type req: C{PreparedRequest}
        @todo: Should I bother filling `response.headers` and processing
               If-Modified-Since and friends using `os.stat`?
        """
        path = os.path.normcase(os.path.normpath(url2pathname(req.path_url)))
        response = requests.Response()

        response.status_code, response.reason = self._chkpath(req.method, path)
        if response.status_code == 200 and req.method.lower() != 'head':
            try:
                response.raw = open(path, 'rb')
            except (OSError, IOError) as err:
                response.status_code = 500
                response.reason = str(err)

        if isinstance(req.url, bytes):
            response.url = req.url.decode('utf-8')
        else:
            response.url = req.url

        response.request = req
        response.connection = self

        return response

    def close(self):
        pass

(Несмотря на название, он был полностью написан до того, как я решил проверить Google, поэтому он не имеет ничего общего с b1r3k.) Как и в случае с другим ответом, выполните следующие действия:

requests_session = requests.session()
requests_session.mount('file://', LocalFileAdapter())
r = requests_session.get('file:///path/to/your/file')

Ответ 3

packages/urllib3/poolmanager.py в значительной степени объясняет это. Запросы не поддерживают локальный URL.

pool_classes_by_scheme = {                                                        
    'http': HTTPConnectionPool,                                                   
    'https': HTTPSConnectionPool,                                              
}                                                                                 

Ответ 4

В недавнем проекте у меня была та же проблема. Поскольку запросы не поддерживают схему "файл", я буду исправлять наш код для загрузки содержимого локально. Во-первых, я определяю функцию для замены requests.get:

def local_get(self, url):
    "Fetch a stream from local files."
    p_url = six.moves.urllib.parse.urlparse(url)
    if p_url.scheme != 'file':
        raise ValueError("Expected file scheme")

    filename = six.moves.urllib.request.url2pathname(p_url.path)
    return open(filename, 'rb')

Затем, где-то в тестовой установке или декорировании тестовой функции, я использую mock.patch для исправления функции get по запросам:

@mock.patch('requests.get', local_get)
def test_handle_remote_file(self):
    ...

Этот метод несколько хрупкий - это не помогает, если базовый код вызывает requests.request или создает Session и вызывает это. Может быть способ исправления запросов на более низком уровне для поддержки file: URL-адресов, но в моем первоначальном исследовании, похоже, не было очевидной точки привязки, поэтому я пошел с этим более простым подходом.