qgreenland/util/misc.py
def fetch_and_write_file(url, *, output_dir, session=None, verify=True):  # noqa:C901
    """Attempt to download and write file from url.

    Assumes filename from URL or content-disposition header.
    """
    if url.startswith('ftp://'):
        if not verify:
            raise RuntimeError(
                'Ignoring TLS certificate verification is not supported for FTP sources.'
            )

        _ftp_fetch_and_write(url, output_dir)
    else:
        # TODO: Share the session across requests somehow?
        if not session:
            session = create_earthdata_authenticated_session(hosts=[url], verify=verify)

        with session.get(url, timeout=REQUEST_TIMEOUT, stream=True) as resp:

            # Try to extract the filename from the `content-disposition` header
            if (
                (disposition := resp.headers.get('content-disposition'))
                and 'filename' in disposition
            ):
                # Sometimes the filename is quoted, sometimes it's not.
                parsed = cgi.parse_header(disposition)
                # Handle case where disposition itself (usually "attachment")
                # isn't present (geothermal heat flux :bell:).
                if 'filename' in parsed[0]:
                    fn = re.match(
                        'filename="?(.*)"?',
                        parsed[0]
                    ).groups()[0].strip('\'"')
                else:
                    fn = parsed[1]['filename']
            else:
                if not (fn := _filename_from_url(url)):
                    raise RuntimeError(
                        f'Failed to retrieve output filename from {url}'
                    )

            fp = os.path.join(output_dir, fn)

            if resp.status_code != 200:
                raise RuntimeError(
                    f"Received '{resp.status_code}' from {resp.request.url}."
                    f'Content: {resp.text}'
                )

            with open(fp, 'wb') as f:
                for chunk in resp.iter_content(chunk_size=CHUNK_SIZE):
                    f.write(chunk)

        return fp
Attempt to download and write file from url.

    Assumes filename from URL or content-disposition header.
    
Attempt to download and write file from url.
Assumes filename from URL or content-disposition header.
Attempt to download and write file from url . Assumes filename from URL or content - disposition header .
def fetch_and_write_file(url, *, output_dir, session=None, verify=True):  
    if url.startswith('ftp://'):
        if not verify:
            raise RuntimeError(
                'Ignoring TLS certificate verification is not supported for FTP sources.'
            )
        _ftp_fetch_and_write(url, output_dir)
    else:
        if not session:
            session = create_earthdata_authenticated_session(hosts=[url], verify=verify)
        with session.get(url, timeout=REQUEST_TIMEOUT, stream=True) as resp:
            if (
                (disposition := resp.headers.get('content-disposition'))
                and 'filename' in disposition
            ):
                parsed = cgi.parse_header(disposition)
                if 'filename' in parsed[0]:
                    fn = re.match(
                        'filename="?(.*)"?',
                        parsed[0]
                    ).groups()[0].strip('\'"')
                else:
                    fn = parsed[1]['filename']
            else:
                if not (fn := _filename_from_url(url)):
                    raise RuntimeError(
                        f'Failed to retrieve output filename from {url}'
                    )
            fp = os.path.join(output_dir, fn)
            if resp.status_code != 200:
                raise RuntimeError(
                    f"Received '{resp.status_code}' from {resp.request.url}."
                    f'Content: {resp.text}'
                )
            with open(fp, 'wb') as f:
                for chunk in resp.iter_content(chunk_size=CHUNK_SIZE):
                    f.write(chunk)
        return fp
def fetch_and_write_file ( url , * , output_dir , session = None , verify = True ) : if url . startswith ( 'ftp://' ) : if not verify : raise RuntimeError ( 'Ignoring TLS certificate verification is not supported for FTP sources.' ) _ftp_fetch_and_write ( url , output_dir ) else : if not session : session = create_earthdata_authenticated_session ( hosts = [ url ] , verify = verify ) with session . get ( url , timeout = REQUEST_TIMEOUT , stream = True ) as resp : if ( ( disposition := resp . headers . get ( 'content-disposition' ) ) and 'filename' in disposition ) : parsed = cgi . parse_header ( disposition ) if 'filename' in parsed [ 0 ] : fn = re . match ( 'filename="?(.*)"?' , parsed [ 0 ] ) . groups ( ) [ 0 ] . strip ( '\'"' ) else : fn = parsed [ 1 ] [ 'filename' ] else : if not ( fn := _filename_from_url ( url ) ) : raise RuntimeError ( f'Failed to retrieve output filename from {url}' ) fp = os . path . join ( output_dir , fn ) if resp . status_code != 200 : raise RuntimeError ( f"Received '{resp.status_code}' from {resp.request.url}." f'Content: {resp.text}' ) with open ( fp , 'wb' ) as f : for chunk in resp . iter_content ( chunk_size = CHUNK_SIZE ) : f . write ( chunk ) return fp
Attempt to download and write file from url.
Attempt to download and write file from url .
# noqa:C901 """Attempt to download and write file from url.

    Assumes filename from URL or content-disposition header.
    """ # TODO: Share the session across requests somehow? # Try to extract the filename from the `content-disposition` header # Sometimes the filename is quoted, sometimes it's not. # Handle case where disposition itself (usually "attachment") # isn't present (geothermal heat flux :bell:).
url
output_dir
session
verify