opal_client/policy/fetcher.py
async def _fetch_policy_bundle(
        self,
        directories: List[str] = ['.'],
        base_hash: Optional[str] = None
    ) -> Optional[PolicyBundle]:
        """
        Fetches the bundle. May throw, in which case we retry again.
        """
        params = {"path": directories}
        if base_hash is not None:
            params["base_hash"] = base_hash
        async with aiohttp.ClientSession() as session:
            try:
                async with session.get(
                    self._policy_endpoint_url,
                    headers={'content-type': 'text/plain', **self._auth_headers},
                    params=params
                ) as response:
                    if response.status == status.HTTP_404_NOT_FOUND:
                        logger.warning("requested paths not found: {paths}", paths=directories)
                        raise HTTPException(
                            status_code=status.HTTP_404_NOT_FOUND,
                            detail=f"requested path {self._policy_endpoint_url} was not found in the policy repo!"
                        )


                    # may throw ValueError
                    await throw_if_bad_status_code(response, expected=[status.HTTP_200_OK])

                    # may throw Validation Error
                    bundle = await response.json()
                    return force_valid_bundle(bundle)
            except aiohttp.ClientError as e:
                logger.warning("server connection error: {err}", err=repr(e))
                raise

        Fetches the bundle. May throw, in which case we retry again.
        
Fetches the bundle. May throw, in which case we retry again.
Fetches the bundle . May throw in which case we retry again .
async def _fetch_policy_bundle(
        self,
        directories: List[str] = ['.'],
        base_hash: Optional[str] = None
    ) -> Optional[PolicyBundle]:
        params = {"path": directories}
        if base_hash is not None:
            params["base_hash"] = base_hash
        async with aiohttp.ClientSession() as session:
            try:
                async with session.get(
                    self._policy_endpoint_url,
                    headers={'content-type': 'text/plain', **self._auth_headers},
                    params=params
                ) as response:
                    if response.status == status.HTTP_404_NOT_FOUND:
                        logger.warning("requested paths not found: {paths}", paths=directories)
                        raise HTTPException(
                            status_code=status.HTTP_404_NOT_FOUND,
                            detail=f"requested path {self._policy_endpoint_url} was not found in the policy repo!"
                        )
                    await throw_if_bad_status_code(response, expected=[status.HTTP_200_OK])
                    bundle = await response.json()
                    return force_valid_bundle(bundle)
            except aiohttp.ClientError as e:
                logger.warning("server connection error: {err}", err=repr(e))
                raise
async def _fetch_policy_bundle ( self , directories : List [ str ] = [ '.' ] , base_hash : Optional [ str ] = None ) -> Optional [ PolicyBundle ] : params = { "path" : directories } if base_hash is not None : params [ "base_hash" ] = base_hash async with aiohttp . ClientSession ( ) as session : try : async with session . get ( self . _policy_endpoint_url , headers = { 'content-type' : 'text/plain' , ** self . _auth_headers } , params = params ) as response : if response . status == status . HTTP_404_NOT_FOUND : logger . warning ( "requested paths not found: {paths}" , paths = directories ) raise HTTPException ( status_code = status . HTTP_404_NOT_FOUND , detail = f"requested path {self._policy_endpoint_url} was not found in the policy repo!" ) await throw_if_bad_status_code ( response , expected = [ status . HTTP_200_OK ] ) bundle = await response . json ( ) return force_valid_bundle ( bundle ) except aiohttp . ClientError as e : logger . warning ( "server connection error: {err}" , err = repr ( e ) ) raise
Fetches the bundle.
Fetches the bundle .
"""
        Fetches the bundle. May throw, in which case we retry again.
        """ # may throw ValueError # may throw Validation Error
self
directories
List[str]
base_hash
Optional[str]