Given a URL, look for the corresponding file in the local cache. If it's not there, download it. Then return the path to the cached file. Return: None in case of non-recoverable file (non-existent or inaccessible url + no cache on disk). Local path (string) otherwise
(
url,
cache_dir=None,
force_download=False,
proxies=None,
etag_timeout=10,
resume_download=False,
user_agent: Union[Dict, str, None] = None,
local_files_only=False,
)
| 648 | |
| 649 | |
| 650 | def get_from_cache( |
| 651 | url, |
| 652 | cache_dir=None, |
| 653 | force_download=False, |
| 654 | proxies=None, |
| 655 | etag_timeout=10, |
| 656 | resume_download=False, |
| 657 | user_agent: Union[Dict, str, None] = None, |
| 658 | local_files_only=False, |
| 659 | ) -> Optional[str]: |
| 660 | """ |
| 661 | Given a URL, look for the corresponding file in the local cache. |
| 662 | If it's not there, download it. Then return the path to the cached file. |
| 663 | |
| 664 | Return: |
| 665 | None in case of non-recoverable file (non-existent or inaccessible url + no cache on disk). |
| 666 | Local path (string) otherwise |
| 667 | """ |
| 668 | if cache_dir is None: |
| 669 | cache_dir = TRANSFORMERS_CACHE |
| 670 | if isinstance(cache_dir, Path): |
| 671 | cache_dir = str(cache_dir) |
| 672 | |
| 673 | os.makedirs(cache_dir, exist_ok=True) |
| 674 | |
| 675 | etag = None |
| 676 | if not local_files_only: |
| 677 | try: |
| 678 | response = requests.head(url, allow_redirects=True, proxies=proxies, timeout=etag_timeout) |
| 679 | if response.status_code == 200: |
| 680 | etag = response.headers.get("ETag") |
| 681 | except (EnvironmentError, requests.exceptions.Timeout): |
| 682 | # etag is already None |
| 683 | pass |
| 684 | |
| 685 | filename = url_to_filename(url, etag) |
| 686 | |
| 687 | # get cache path to put the file |
| 688 | cache_path = os.path.join(cache_dir, filename) |
| 689 | |
| 690 | # etag is None = we don't have a connection, or url doesn't exist, or is otherwise inaccessible. |
| 691 | # try to get the last downloaded one |
| 692 | if etag is None: |
| 693 | if os.path.exists(cache_path): |
| 694 | return cache_path |
| 695 | else: |
| 696 | matching_files = [ |
| 697 | file |
| 698 | for file in fnmatch.filter(os.listdir(cache_dir), filename + ".*") |
| 699 | if not file.endswith(".json") and not file.endswith(".lock") |
| 700 | ] |
| 701 | if len(matching_files) > 0: |
| 702 | return os.path.join(cache_dir, matching_files[-1]) |
| 703 | else: |
| 704 | # If files cannot be found and local_files_only=True, |
| 705 | # the models might've been found if local_files_only=False |
| 706 | # Notify the user about that |
| 707 | if local_files_only: |
no test coverage detected