Python istekleri yerel bir URL'den bir dosya getir

Question

Python'un requests kitaplığımı uygulamamın bir yönteminde kullanıyorum. Yöntemin vücut gibidir:

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)) 

: Ben ne yapmak istiyorum gibi sahte yerel url geçmektir Ancak bu yöntem için bazı birim testleri yazmak istiyorum

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

yerel bir uRL'ye sahip requests.get diyoruz , ben istisna aşağıda KeyError var:

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' 

Soru, yerel bir URL'yi requests.get'a nasıl geçirebilirim?

Not: Yukarıdaki örneği oluşur. Muhtemelen birçok hata içerir.

Answer 1

@WooParadog açıkladı istekleri kitaplığı yerel dosyaları nasıl ele alacağını bilmiyor. Bununla birlikte, güncel sürüm transport adapters tanımlamasına izin verir.

nedenle sadece

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>') 

Ben yukarıdaki örnekte requests-testadapter modülü kullanıyorum .:, yerel dosyaları işlemek mümkün olacaktır adaptörü sahibi örn tanımlayabilirsiniz.

Answer 2

packages/urllib3/poolmanager.py hemen hemen her şeyi açıklıyor. İstekler yerel URL’yi desteklemiyor.

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

Answer 3

Son bir projede, aynı sorunu yaşadım. İstekler "dosya" şemasını desteklemediğinden, içeriği yerel olarak yüklemek için kodumuzu ekleyeceğiz. yere deney düzeneğinde, Sonra

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') 

veya test fonksiyonu dekorasyon, ben isteklerde olsun fonksiyonunu yama mock.patch kullanın: Birincisi, requests.get değiştirmek için bir işlev tanımlamak

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

Bu tekniktir Biraz kırılgan - Alttaki kod requests.request'u çağırırsa veya Session'u yapılandırırsa ve bunu çağırırsa bu yardımcı olmaz. file: URL'lerini desteklemek için istekleri daha düşük bir seviyeye getirmenin bir yolu olabilir, ancak ilk araştırmamda bariz bir kanca noktası görünmüyordu, bu yüzden daha basit bir yaklaşımla gittim.

Answer 4

Bu yazıda b1r3k'inkinden daha elverişli olan ve İsteklerin kendisinden başka ek bağımlılıkları olmayan bir yazım adaptörü var. Henüz fazla test etmedim, ama denediğim şey hatasız görünüyor.

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 

(. Adı rağmen tamamen Google'ı kontrol etmeyi düşünmemiş önce yazdığı, bu nedenle b1r3k yıllardan ile ilgisi yoktur edildi) diğer cevap olduğu gibi, bu izleyin:

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