Zy Filemanager

📁 File Browser

Name	Size	Permissions	Actions
📁 __pycache__	DIR	0755	🗑️
📁 caches	DIR	0755	🗑️
📄 __init__.py (465 bytes)	465 B	0644	✍️ ⬇️ 🗑️
📄 _cmd.py (1,379 bytes)	1.35 KB	0644	✍️ ⬇️ 🗑️
📄 adapter.py (5,033 bytes)	4.92 KB	0644	✍️ ⬇️ 🗑️
📄 cache.py (1,535 bytes)	1.5 KB	0644	✍️ ⬇️ 🗑️
📄 compat.py (778 bytes)	778 B	0644	✍️ ⬇️ 🗑️
📄 controller.py (16,416 bytes)	16.03 KB	0644	✍️ ⬇️ 🗑️
📄 filewrapper.py (3,946 bytes)	3.85 KB	0644	✍️ ⬇️ 🗑️
📄 heuristics.py (4,154 bytes)	4.06 KB	0644	✍️ ⬇️ 🗑️
📄 serialize.py (7,105 bytes)	6.94 KB	0644	✍️ ⬇️ 🗑️
📄 wrapper.py (774 bytes)	774 B	0644	✍️ ⬇️ 🗑️

📤 Upload Files

Single File

Multiple Files

ZIP Extract

ZIP file will be deleted after extraction

➕ Create New

New Folder

New File

✍️ Editing: adapter.py

            File: /opt/alt/python38/lib/python3.8/site-packages/pip/_vendor/cachecontrol/adapter.py

            Size: 4.92 KB | 
            Last Modified: 2023-11-14 04:40:26        

# SPDX-FileCopyrightText: 2015 Eric Larson
#
# SPDX-License-Identifier: Apache-2.0

import types
import functools
import zlib

from pip._vendor.requests.adapters import HTTPAdapter

from .controller import CacheController, PERMANENT_REDIRECT_STATUSES
from .cache import DictCache
from .filewrapper import CallbackFileWrapper

class CacheControlAdapter(HTTPAdapter):
    invalidating_methods = {"PUT", "PATCH", "DELETE"}

def __init__(
        self,
        cache=None,
        cache_etags=True,
        controller_class=None,
        serializer=None,
        heuristic=None,
        cacheable_methods=None,
        *args,
        **kw
    ):
        super(CacheControlAdapter, self).__init__(*args, **kw)
        self.cache = DictCache() if cache is None else cache
        self.heuristic = heuristic
        self.cacheable_methods = cacheable_methods or ("GET",)

controller_factory = controller_class or CacheController
        self.controller = controller_factory(
            self.cache, cache_etags=cache_etags, serializer=serializer
        )

def send(self, request, cacheable_methods=None, **kw):
        """
        Send a request. Use the request information to see if it
        exists in the cache and cache the response if we need to and can.
        """
        cacheable = cacheable_methods or self.cacheable_methods
        if request.method in cacheable:
            try:
                cached_response = self.controller.cached_request(request)
            except zlib.error:
                cached_response = None
            if cached_response:
                return self.build_response(request, cached_response, from_cache=True)

# check for etags and add headers if appropriate
            request.headers.update(self.controller.conditional_headers(request))

resp = super(CacheControlAdapter, self).send(request, **kw)

return resp

def build_response(
        self, request, response, from_cache=False, cacheable_methods=None
    ):
        """
        Build a response by making a request or using the cache.

This will end up calling send and returning a potentially
        cached response
        """
        cacheable = cacheable_methods or self.cacheable_methods
        if not from_cache and request.method in cacheable:
            # Check for any heuristics that might update headers
            # before trying to cache.
            if self.heuristic:
                response = self.heuristic.apply(response)

# apply any expiration heuristics
            if response.status == 304:
                # We must have sent an ETag request. This could mean
                # that we've been expired already or that we simply
                # have an etag. In either case, we want to try and
                # update the cache if that is the case.
                cached_response = self.controller.update_cached_response(
                    request, response
                )

if cached_response is not response:
                    from_cache = True

# We are done with the server response, read a
                # possible response body (compliant servers will
                # not return one, but we cannot be 100% sure) and
                # release the connection back to the pool.
                response.read(decode_content=False)
                response.release_conn()

response = cached_response

# We always cache the 301 responses
            elif int(response.status) in PERMANENT_REDIRECT_STATUSES:
                self.controller.cache_response(request, response)
            else:
                # Wrap the response file with a wrapper that will cache the
                #   response when the stream has been consumed.
                response._fp = CallbackFileWrapper(
                    response._fp,
                    functools.partial(
                        self.controller.cache_response, request, response
                    ),
                )
                if response.chunked:
                    super_update_chunk_length = response._update_chunk_length

def _update_chunk_length(self):
                        super_update_chunk_length()
                        if self.chunk_left == 0:
                            self._fp._close()

response._update_chunk_length = types.MethodType(
                        _update_chunk_length, response
                    )

resp = super(CacheControlAdapter, self).build_response(request, response)

# See if we should invalidate the cache.
        if request.method in self.invalidating_methods and resp.ok:
            cache_url = self.controller.cache_url(request.url)
            self.cache.delete(cache_url)

# Give the request a from_cache attr to let people use it
        resp.from_cache = from_cache