Python http.client 模块，NOT_FOUND 实例源码

def save(self):
        try:
            self.create()

        except K8sApiException as e:
            if e.status == client.NOT_FOUND:
                logging.warning("Third Party Resource is not ready yet. "
                                "Report will be skipped")
                return
            if e.status == client.METHOD_NOT_ALLOWED:
                logging.error("API is blocked. Report will be skipped")
                return
            if e.status != client.CONFLICT:
                raise e
            logging.info("Previous resource has been detected. Recreating...")

            try:
                self.remove()
            except K8sApiException as e:
                if e.status != client.NOT_FOUND:
                    raise e

            self.create()

def save(self):
        """Save custom object if not exists"""
        try:
            self.create()

        except K8sApiException as e:
            if e.status == client.NOT_FOUND:
                logging.warning("Custom Resource Definition is not ready yet. "
                                "Report will be skipped")
                return
            if e.status == client.METHOD_NOT_ALLOWED:
                logging.error("API is blocked. Report will be skipped")
                return
            if e.status != client.CONFLICT:
                raise e
            logging.warning("Previous definition has been detected. "
                            "Recreating...")

            try:
                self.remove()
            except K8sApiException as e:
                if e.status != client.NOT_FOUND:
                    raise e

            self.create()

def _download_wrapped_file(download):
    download_path = download.abspath
    # We do not allow symlinks as downloads for security reasons
    if not path.exists(download_path) or path.islink(download_path):
        return HttpResponse("Download not found", status=HTTP_NOT_FOUND)
    wrapper = FileWrapper(open(download_path, "rb"))
    response = HttpResponse(wrapper, content_type='application/force-download')
    response['Content-Disposition'] = 'attachment; filename="{}"'.format(
        DOWNLOAD_FNAME_TEMLATE.format(
            filename=path.basename(download_path),
            download_pk=download.pk,
            problem_slug=download.problem.slug
        )
    )
    response['Content-Length'] = path.getsize(download_path)
    return response

def test_check_resp_status_and_retry_image_not_found(self):
        mock_resp_badgateway = mock.Mock()
        mock_resp_badgateway.status = httplib.NOT_FOUND
        self.glance.XenAPI.Failure = FakeXenAPIException
        self.assertRaises(
            self.glance.XenAPI.Failure,
            self.glance.check_resp_status_and_retry,
            mock_resp_badgateway,
            'fake_image_id',
            'fake_url')

def __get_task_response(self, task_id):
        """ Get task response by task id.

        Args:
            task_id (str): id of the task.

        Returns:
            task_response (AugmentedTaskDTO) of the task.
        """

        retry_interval = 1
        retries = self.GET_TASK_MAX_RETRIES
        for retry in range(retries):
            try:
                task_result = self.nb_api.task.getTask(taskId=task_id)
                task_response = task_result.response
                return task_response
            except HTTPError as err:
                error_code = err.response.status_code
                error_result = err.response._content.decode()
                if error_code == NOT_FOUND:
                    if retry < retries - 1:
                        time.sleep(retry_interval)
                        retry_interval *= self.EXPONENTIAL_BACKOFF_MULTIPLIER
                    else:
                        raise ApiClientException(
                            "Max retries ({0}) exceeded\nHTTP error code: {1}\nerror result: {2}"
                            .format(retries, error_code, json.loads(error_result)))
                else:
                    raise ApiClientException("HTTP error code: {0}\nerror result: {1}"
                                             .format(error_code, json.loads(error_result)))
        raise ApiClientException("Didn't get any valid task response for '{}'".format(task_id))

def exists(self, namespace="default"):
        header_params = {
            'Content-Type': "application/json",
            'Accept': "application/json"
        }

        auth_settings = ['BearerToken']

        resource_path = "/".join([
            "/apis",
            self.type_url,
            self.type_version,
            "namespaces", namespace,
            self.type_name + "s"
        ])

        try:
            self.api.api_client.call_api(
                resource_path,
                'GET',
                header_params,
                auth_settings=auth_settings)

        except K8sApiException as e:
            if e.status == client.CONFLICT or e.status == client.NOT_FOUND:
                return False
            raise e

        return True

def exists(self, namespace="default"):
        """Check if custom resource definition exists"""
        try:
            self.api.api_client.call_api(
                self.resource_path_crd_type,
                'GET',
                self.header_params,
                auth_settings=self.auth_settings)

        except K8sApiException as e:
            if e.status == client.CONFLICT or e.status == client.NOT_FOUND:
                return False
            raise e
        return True

def test_post_item_pictures__wrong_item_uuid(self):
        resp = self.app.post('/items/{item_uuid}/pictures/'.format(
            item_uuid=WRONG_UUID),
            data={'image': (BytesIO(b'my file contents'), 'testimage.jpg')},
            content_type='multipart/form-data')
        assert resp.status_code == client.NOT_FOUND
        assert Picture.select().count() == 0

def post(self, item_uuid):
        """Insert a new picture for the specified item"""
        if 'image' not in request.files:
            return {"message": "No image received"},\
                client.BAD_REQUEST

        try:
            item = Item.get(Item.uuid == item_uuid)
        except Item.DoesNotExist:
            return None, client.NOT_FOUND

        file = request.files['image']
        picture_uuid = uuid.uuid4()
        extension = os.path.splitext(file.filename)[1][1:]

        if extension not in ALLOWED_EXTENSION:
            return {"message": "File extension not allowed"},\
                client.BAD_REQUEST

        utils.save_image(file, picture_uuid, extension)

        return Picture.create(
            uuid=picture_uuid,
            extension=extension,
            item=item
        ).json(), client.CREATED

def get(self, picture_uuid):
        """Retrieve the picture specified by picture_uuid"""
        try:
            picture = Picture.get(Picture.uuid == picture_uuid)
        except Picture.DoesNotExist:
            return None, client.NOT_FOUND

        return send_from_directory(utils.get_image_folder(),
                                   picture.filename, as_attachment=True)

def delete(self, picture_uuid):
        """Remove the picture specified by picture_uuid"""
        try:
            obj = Picture.get(Picture.uuid == picture_uuid)
        except Picture.DoesNotExist:
            return None, client.NOT_FOUND
        obj.delete_instance()
        return None, client.NO_CONTENT

def patch(self, item_uuid):
        """Edit the item specified by item_uuid"""
        try:
            obj = Item.get(Item.uuid == item_uuid)
        except Item.DoesNotExist:
            return None, client.NOT_FOUND

        request_data = request.get_json(force=True)

        errors = Item.validate_input(request_data, partial=True)
        if errors:
            return errors, client.BAD_REQUEST

        data = request_data['data']['attributes']

        name = data.get('name')
        price = data.get('price')
        description = data.get('description')
        availability = data.get('availability')
        category = data.get('category')

        if name:
            obj.name = name

        if price:
            obj.price = price

        if description:
            obj.description = description

        if availability:
            obj.availability = availability

        if category:
            obj.category = category

        obj.save()

        return generate_response(obj.json(), client.OK)

def download_file(request, download_id):
    download = get_object_or_404(Download, pk=download_id)

    if download.problem.course_set.filter(
            participants__pk=request.user.pk
    ).exists():
            return _download_wrapped_file(download)

    return HttpResponse(
        "Not Found", status=HTTP_NOT_FOUND
    )