Python richiede il recupero di un file da un url locale

Question

Sto usando la libreria Python requests in un metodo della mia applicazione. Il corpo del metodo è simile al seguente:

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

mi piacerebbe scrivere alcuni test di unità per questo metodo, tuttavia, quello che voglio fare è quello di passare un falso URL locale come:

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

Quando chiamo requests.get con un'URL locale, ho avuto l'KeyError eccezione di seguito:

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' 

La domanda è: come posso passare un url locale alle richieste .get?

PS: ho inventato l'esempio precedente. Probabilmente contiene molti errori.

Answer 1

Come @WooParadog ha spiegato che la libreria delle richieste non sa come gestire i file locali. Anche se, la versione corrente consente di definire transport adapters.

Pertanto si può semplicemente definire Possiedi un adattatore che sarà in grado di gestire i file locali, per es .:

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

che sto utilizzando il modulo requests-testadapter nell'esempio di cui sopra.

Answer 2

packages/urllib3/poolmanager.py praticamente lo spiega. Le richieste non supportano l'URL locale.

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

Answer 3

In un progetto recente, ho riscontrato lo stesso problema. Poiché le richieste non supportano lo schema "file", cercherò il nostro codice per caricare localmente il contenuto. In primo luogo, definire una funzione per sostituire 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') 

Poi, da qualche parte nella messa a punto di test o decorare la funzione di test, io uso mock.patch patchare la funzione get sulle richieste:

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

Questa tecnica è un po 'fragile: non aiuta se il codice sottostante chiama requests.request o crea un Session e lo chiama. Potrebbe esserci un modo per correggere le richieste a un livello inferiore per supportare gli URL file:, ma nella mia analisi iniziale, non sembrava esserci un ovvio punto di aggancio, quindi sono andato con questo approccio più semplice.

Answer 4

Ecco un adattatore per il trasporto che ho scritto che è più funzionale di quello di b1r3k e non ha altre dipendenze oltre le richieste stesse. Non l'ho ancora testato in modo esauriente, ma quello che ho provato sembra essere privo di bug.

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 

(. Nonostante il nome, è stato completamente scritto prima ho pensato di controllare Google, quindi non ha nulla a che fare con b1r3k di) Come con l'altra risposta, segui questo con:

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