• 设为首页
  • 点击收藏
  • 手机版
    手机扫一扫访问
    迪恩网络手机版
  • 关注官方公众号
    微信扫一扫关注
    迪恩网络公众号

Python utc.utcnow函数代码示例

原作者: [db:作者] 来自: [db:来源] 收藏 邀请

本文整理汇总了Python中superdesk.utc.utcnow函数的典型用法代码示例。如果您正苦于以下问题:Python utcnow函数的具体用法?Python utcnow怎么用?Python utcnow使用的例子?那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。



在下文中一共展示了utcnow函数的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。

示例1: test_ingest_update

    def test_ingest_update(self):
        provider, provider_service = self.setup_reuters_provider()
        items = provider_service.fetch_ingest(reuters_guid)
        items[0]['ingest_provider'] = provider['_id']
        items[0]['expiry'] = utcnow() + timedelta(hours=11)

        self.ingest_items(items, provider, provider_service)

        self.assertEqual(items[0]['unique_id'], 1)
        original_id = items[0]['_id']

        items = provider_service.fetch_ingest(reuters_guid)
        items[0]['ingest_provider'] = provider['_id']
        items[0]['expiry'] = utcnow() + timedelta(hours=11)
        # change the headline
        items[0]['headline'] = 'Updated headline'

        # ingest the item again
        self.ingest_items(items, provider, provider_service)

        # see the update to the headline and unique_id survives
        elastic_item = self.app.data._search_backend('ingest').find_one('ingest', _id=original_id, req=None)
        self.assertEqual(elastic_item['headline'], 'Updated headline')
        self.assertEqual(elastic_item['unique_id'], 1)
        self.assertEqual(elastic_item['unique_name'], '#1')
开发者ID:liveblog,项目名称:superdesk-core,代码行数:25,代码来源:update_ingest_tests.py


示例2: test_ingest_update

    def test_ingest_update(self):
        provider_name = 'reuters'
        guid = 'tag_reuters.com_2014_newsml_KBN0FL0NN:5'
        provider = get_resource_service('ingest_providers').find_one(name=provider_name, req=None)
        provider_service = self._get_provider_service(provider)
        provider_service.provider = provider
        provider_service.URL = provider.get('config', {}).get('url')
        items = provider_service.fetch_ingest(guid)
        items[0]['ingest_provider'] = provider['_id']
        items[0]['expiry'] = utcnow() + timedelta(hours=11)

        self.ingest_items(items, provider, provider_service)

        self.assertEqual(items[0]['unique_id'], 1)
        original_id = items[0]['_id']

        items = provider_service.fetch_ingest(guid)
        items[0]['ingest_provider'] = provider['_id']
        items[0]['expiry'] = utcnow() + timedelta(hours=11)
        # change the headline
        items[0]['headline'] = 'Updated headline'

        # ingest the item again
        self.ingest_items(items, provider, provider_service)

        # see the update to the headline and unique_id survives
        elastic_item = self.app.data._search_backend('ingest').find_one('ingest', _id=original_id, req=None)
        self.assertEqual(elastic_item['headline'], 'Updated headline')
        self.assertEqual(elastic_item['unique_id'], 1)
        self.assertEqual(elastic_item['unique_name'], '#1')
开发者ID:MiczFlor,项目名称:superdesk-core,代码行数:30,代码来源:update_ingest_tests.py


示例3: test_new_zealand_content_with_source_NZN

    def test_new_zealand_content_with_source_NZN(self):
        article = {
            'source': 'NZN',
            'anpa_category': [{'qcode': 's'}],
            'headline': 'This is a test headline',
            'byline': 'joe',
            'slugline': 'slugline',
            'subject': [{'qcode': '15017000'}],
            'anpa_take_key': 'take_key',
            'unique_id': '1',
            'type': 'text',
            'format': 'preserved',
            'body_html': 'The story body',
            'word_count': '1',
            'priority': '1',
            'firstcreated': utcnow(),
            'versioncreated': utcnow(),
            'lock_user': ObjectId(),
            'task': {
                'desk': self.desks[1][config.ID_FIELD]
            }
        }

        subscriber = self.app.data.find('subscribers', None, None)[0]
        seq, item = self._formatter.format(article, subscriber)[0]
        item = json.loads(item)
        self.assertGreater(int(seq), 0)
        test_article = json.loads(item.get('data'))
        self.assertEqual(test_article['source'], 'NZN')
开发者ID:akintolga,项目名称:superdesk-aap,代码行数:29,代码来源:aap_bulletinbuilder_formatter_tests.py


示例4: setUp

 def setUp(self):
     try:
         from superdesk.publish.publish_content import get_queue_items
     except ImportError:
         self.fail("Could not import function under test (transmit_items).")
     else:
         self.func_under_test = get_queue_items
         self.queue_items = [
             {'_id': ObjectId(), 'state': 'pending', 'item_id': 'item_1', 'item_version': 4,
              'headline': 'pending headline', 'destination': {}},
             {'_id': ObjectId(), 'state': 'retrying', 'item_id': 'item_2', 'item_version': 4,
              'headline': 'retrying headline', 'retry_attempt': 2,
              'next_retry_attempt_at': utcnow() + timedelta(minutes=30)},
             {'_id': ObjectId(), 'state': 'success', 'item_id': 'item_3', 'item_version': 4,
              'headline': 'success headline', 'retry_attempt': 4,
              'next_retry_attempt_at': utcnow() + timedelta(minutes=-30)},
             {'_id': ObjectId(), 'state': 'failed', 'item_id': 'item_4', 'item_version': 4,
              'headline': 'failed headline', 'retry_attempt': 10,
              'next_retry_attempt_at': utcnow() + timedelta(minutes=-30)},
             {'_id': ObjectId(), 'state': 'canceled', 'item_id': 'item_5', 'item_version': 4,
              'headline': 'canceled headline', 'retry_attempt': 4,
              'next_retry_attempt_at': utcnow() + timedelta(minutes=-30)},
             {'_id': ObjectId(), 'state': 'retrying', 'item_id': 'item_6', 'item_version': 4,
              'headline': 'retrying headline', 'retry_attempt': 2,
              'next_retry_attempt_at': utcnow() + timedelta(minutes=-30)},
         ]
         self.app.data.insert('publish_queue', self.queue_items)
开发者ID:hlmnrmr,项目名称:superdesk-core,代码行数:27,代码来源:publish_content_tests.py


示例5: on_update

 def on_update(self, updates, original):
     # in the case we have a comment
     if original['post_status'] == 'comment':
         original['blog'] = original['groups'][1]['refs'][0]['item']['client_blog']
         updates['blog'] = original['groups'][1]['refs'][0]['item']['client_blog']
         # if the length of the comment is not between 1 and 300 then we get an error
         check_comment_length(original['groups'][1]['refs'][0]['item']['text'])
     # check if updates `content` is diffrent then the original.
     content_diff = False
     if not updates.get('groups', False):
         content_diff = False
     elif len(original['groups'][1]['refs']) != len(updates['groups'][1]['refs']):
         content_diff = True
     else:
         for index, val in enumerate(updates['groups'][1]['refs']):
             item = get_resource_service('archive').find_one(req=None, _id=val['residRef'])
             if item['text'] != original['groups'][1]['refs'][index]['item']['text']:
                 content_diff = True
                 break
     if(content_diff):
         updates['content_updated_date'] = utcnow()
     # check permission
     post = original.copy()
     post.update(updates)
     self.check_post_permission(post)
     # when publishing, put the published item from drafts and contributions at the top of the timeline
     if updates.get('post_status') == 'open' and original.get('post_status') in ('draft', 'submitted', 'comment'):
         updates['order'] = self.get_next_order_sequence(original.get('blog'))
         # if you publish a post it will save a published date and register who did it
         updates['published_date'] = utcnow()
         updates['publisher'] = getattr(flask.g, 'user', None)
     # when unpublishing
     if original.get('post_status') == 'open' and updates.get('post_status') != 'open':
         updates['unpublished_date'] = utcnow()
     super().on_update(updates, original)
开发者ID:wiederspielwert,项目名称:liveblog,代码行数:35,代码来源:posts.py


示例6: filter_expired_items

def filter_expired_items(provider, items):
    try:
        days_to_keep_content = provider.get('days_to_keep', DAYS_TO_KEEP)
        expiration_date = utcnow() - timedelta(days=days_to_keep_content)
        return [item for item in items if item.get('versioncreated', utcnow()) > expiration_date]
    except Exception as ex:
        raise ProviderError.providerFilterExpiredContentError(ex, provider)
开发者ID:petrjasek,项目名称:superdesk-server,代码行数:7,代码来源:update_ingest.py


示例7: setUp

    def setUp(self):
        self.req = ParsedRequest()
        with self.app.test_request_context(self.app.config.get('URL_PREFIX')):
            self.articles = [{'_id': '1', 'urgency': 1, 'headline': 'story', 'state': 'fetched'},
                             {'_id': '2', 'headline': 'prtorque', 'state': 'fetched'},
                             {'_id': '3', 'urgency': 3, 'state': 'fetched', 'flags': {'marked_for_sms': True}},
                             {'_id': '4', 'urgency': 4, 'state': 'fetched', 'task': {'desk': '1'},
                              'ingest_provider': '1'},
                             {'_id': '5', 'urgency': 2, 'state': 'fetched', 'task': {'desk': '2'}, 'priority': 3},
                             {'_id': '6', 'state': 'fetched', 'embargo': utcnow(),
                             'schedule_settings': {'utc_embargo': utcnow() + timedelta(minutes=20)}},
                             {'_id': '7', 'genre': [{'name': 'Sidebar'}], 'state': 'fetched'},
                             {'_id': '8', 'subject': [{'name': 'adult education',
                                                       'qcode': '05001000',
                                                       'parent': '05000000'},
                                                      {'name': 'high schools',
                                                       'qcode': '05005003',
                                                       'parent': '05005000'}], 'state': 'fetched'},
                             {'_id': '9', 'state': 'fetched', 'anpa_category':
                                 [{'qcode': 'a', 'name': 'Aus News'}]},
                             {'_id': '10', 'body_html': '<p>Mention<p>', 'embargo': utcnow(),
                             'schedule_settings': {'utc_embargo': utcnow() - timedelta(minutes=20)}},
                             {'_id': '11', 'place': [{'qcode': 'NSW', 'name': 'NSW'}], 'state': 'fetched'}]

            self.app.data.insert('archive', self.articles)

            self.app.data.insert('filter_conditions',
                                 [{'_id': 1,
                                   'field': 'headline',
                                   'operator': 'like',
                                   'value': 'tor',
                                   'name': 'test-1'}])
            self.app.data.insert('filter_conditions',
                                 [{'_id': 2,
                                   'field': 'urgency',
                                   'operator': 'in',
                                   'value': '2',
                                   'name': 'test-2'}])
            self.app.data.insert('filter_conditions',
                                 [{'_id': 3,
                                   'field': 'urgency',
                                   'operator': 'in',
                                   'value': '3,4,5',
                                   'name': 'test-2'}])
            self.app.data.insert('filter_conditions',
                                 [{'_id': 4,
                                   'field': 'urgency',
                                   'operator': 'nin',
                                   'value': '1,2,3',
                                   'name': 'test-2'}])
            self.app.data.insert('filter_conditions',
                                 [{'_id': 5,
                                   'field': 'urgency',
                                   'operator': 'in',
                                   'value': '2,5',
                                   'name': 'test-2'}])
            self.app.data.insert('content_filters',
                                 [{"_id": 1,
                                   "content_filter": [{"expression": {"fc": [1]}}],
                                   "name": "soccer-only"}])
开发者ID:jerome-poisson,项目名称:superdesk-core,代码行数:60,代码来源:tests.py


示例8: test_auto_publish_without_abstract_other_source

    def test_auto_publish_without_abstract_other_source(self):
        article = {
            'source': 'AAP',
            'anpa_category': [{'qcode': 'c'}],
            'headline': 'This is a test headline',
            'auto_publish': True,
            'byline': 'joe',
            'slugline': 'slugline',
            'subject': [{'qcode': '15017000'}],
            'anpa_take_key': 'take_key',
            'unique_id': '1',
            'type': 'text',
            'format': 'HTML',
            'body_html': 'Sydney, AAP - The story body text.',
            'word_count': '1',
            'priority': '1',
            'firstcreated': utcnow(),
            'versioncreated': utcnow(),
            'lock_user': ObjectId(),
            'task': {
                'desk': self.desks[0][config.ID_FIELD]
            }
        }

        subscriber = self.app.data.find('subscribers', None, None)[0]
        seq, item = self._formatter.format(article, subscriber)[0]
        item = json.loads(item)
        self.assertGreater(int(seq), 0)
        test_article = json.loads(item.get('data'))
        self.assertEqual(test_article['source'], 'AAP')
        self.assertEqual(test_article['abstract'], 'This is a test headline')
        self.assertEqual(test_article['slugline'], 'slugline')
        self.assertEqual(test_article['body_text'], 'Sydney, AAP - The story body text.')
开发者ID:akintolga,项目名称:superdesk-aap,代码行数:33,代码来源:aap_bulletinbuilder_formatter_tests.py


示例9: test_single_category_allow_features

    def test_single_category_allow_features(self):
        article = {
            'source': 'AAP',
            'anpa_category': [{'qcode': 'c'}],
            'headline': 'This is a test headline',
            'byline': 'joe',
            'slugline': 'slugline',
            'subject': [{'qcode': '15017000'}],
            'anpa_take_key': 'take_key',
            'unique_id': '1',
            'type': 'text',
            'format': 'HTML',
            'body_html': 'The story body',
            'word_count': '1',
            'priority': '1',
            'firstcreated': utcnow(),
            'versioncreated': utcnow(),
            'lock_user': ObjectId(),
            'task': {
                'desk': self.desks[0][config.ID_FIELD]
            }
        }

        subscriber = self.app.data.find('subscribers', None, None)[0]
        seq, item = self._formatter.format(article, subscriber)[0]
        item = json.loads(item)
        self.assertGreater(int(seq), 0)
        test_article = json.loads(item.get('data'))
        self.assertEqual(test_article['source'], 'AAP')
        self.assertEqual(test_article['first_category']['qcode'], 'c')
        self.assertEqual(len(test_article['anpa_category']), 1)
        self.assertEqual(test_article['anpa_category'][0]['qcode'], 'c')
开发者ID:akintolga,项目名称:superdesk-aap,代码行数:32,代码来源:aap_bulletinbuilder_formatter_tests.py


示例10: update_provider

def update_provider(provider, rule_set=None, routing_scheme=None):
    """
    Fetches items from ingest provider as per the configuration, ingests them into Superdesk and
    updates the provider.
    """
    if ingest_for_provider_is_already_running(provider):
        return

    try:
        update = {
            LAST_UPDATED: utcnow()
        }

        for items in providers[provider.get('type')].update(provider):
            ingest_items(items, provider, rule_set, routing_scheme)
            stats.incr('ingest.ingested_items', len(items))
            if items:
                update[LAST_ITEM_UPDATE] = utcnow()
        ingest_service = superdesk.get_resource_service('ingest_providers')
        ingest_service.system_update(provider[superdesk.config.ID_FIELD], update, provider)

        if LAST_ITEM_UPDATE not in update and get_is_idle(provider):
            notify_and_add_activity(
                ACTIVITY_EVENT,
                'Provider {{name}} has gone strangely quiet. Last activity was on {{last}}',
                resource='ingest_providers',
                user_list=ingest_service._get_administrators(),
                name=provider.get('name'),
                last=provider[LAST_ITEM_UPDATE].replace(tzinfo=timezone.utc).astimezone(tz=None).strftime("%c"))

        logger.info('Provider {0} updated'.format(provider[superdesk.config.ID_FIELD]))
        push_notification('ingest:update', provider_id=str(provider[superdesk.config.ID_FIELD]))
    finally:
        mark_provider_as_not_running(provider)
开发者ID:yukoff,项目名称:superdesk,代码行数:34,代码来源:update_ingest.py


示例11: test_body_footer

    def test_body_footer(self):
        article = {
            'source': 'AAP',
            'anpa_category': [{'qcode': 's'}],
            'headline': 'This is a test headline',
            'byline': 'joe',
            'slugline': 'slugline',
            'subject': [{'qcode': '15017000'}],
            'anpa_take_key': 'take_key',
            'unique_id': '1',
            'type': 'preformatted',
            'body_html': 'The story body',
            'word_count': '1',
            'priority': '1',
            'firstcreated': utcnow(),
            'versioncreated': utcnow(),
            'lock_user': ObjectId(),
            'body_footer': 'call helpline 999 if you are planning to quit smoking'
        }

        subscriber = self.app.data.find('subscribers', None, None)[0]
        seq, item = self._formatter.format(article, subscriber)[0]

        formatted_article = json.loads(item)
        self.assertEqual(formatted_article['body_text'],
                         'The story body call helpline 999 if you are planning to quit smoking')
开发者ID:actionless,项目名称:superdesk,代码行数:26,代码来源:aap_bulletinbuilder_formatter_tests.py


示例12: TestBulletinBuilderFormatter

    def TestBulletinBuilderFormatter(self):
        article = {
            'source': 'AAP',
            'anpa-category': {'qcode': 'a'},
            'headline': 'This is a test headline',
            'byline': 'joe',
            'slugline': 'slugline',
            'subject': [{'qcode': '02011001'}],
            'anpa_take_key': 'take_key',
            'unique_id': '1',
            'type': 'preformatted',
            'body_html': 'The story body',
            'word_count': '1',
            'priority': '1',
            'firstcreated': utcnow(),
            'versioncreated': utcnow(),
            'lock_user': ObjectId()
        }

        with self.app.app_context():
            subscriber = self.app.data.find('subscribers', None, None)[0]
            f = AAPBulletinBuilderFormatter()
            seq, item = f.format(article, subscriber)[0]
            self.assertGreater(int(seq), 0)
            self.assertEqual(json.dumps(article, default=json_serialize_datetime_objectId), item)
开发者ID:oxcarh,项目名称:superdesk,代码行数:25,代码来源:aap_bulletinbuilder_formatter_tests.py


示例13: TestLocator

    def TestLocator(self):
        article = {
            'source': 'AAP',
            'anpa_category': [{'qcode': 's'}],
            'headline': 'This is a test headline',
            'byline': 'joe',
            'slugline': 'slugline',
            'subject': [{'qcode': '15017000'}],
            'anpa_take_key': 'take_key',
            'unique_id': '1',
            'type': 'preformatted',
            'body_html': 'The story body',
            'word_count': '1',
            'priority': '1',
            'firstcreated': utcnow(),
            'versioncreated': utcnow(),
            'lock_user': ObjectId(),
            'place': [{'qcode': 'VIC', 'name': 'VIC'}]
        }

        subscriber = self.app.data.find('subscribers', None, None)[0]
        f = AAPBulletinBuilderFormatter()
        seq, item = f.format(article, subscriber)[0]
        self.assertGreater(int(seq), 0)
        test_article = json.loads(item)
        self.assertEqual(test_article['headline'], 'This is a test headline')
        self.assertEqual(test_article['place'][0]['qcode'], 'CRIK')
        article['anpa_category'] = [{'qcode': 'a'}]
        article['place'] = [{'qcode': 'VIC', 'name': 'VIC'}]
        seq, item = f.format(article, subscriber)[0]
        self.assertGreater(int(seq), 0)
        test_article = json.loads(item)
        self.assertEqual(test_article['headline'], 'This is a test headline')
        self.assertEqual(test_article['place'][0]['qcode'], 'VIC')
开发者ID:chalkjockey,项目名称:superdesk,代码行数:34,代码来源:aap_bulletinbuilder_formatter_tests.py


示例14: test_files_dont_duplicate_ingest

    def test_files_dont_duplicate_ingest(self):
        provider_name = 'reuters'
        guid = 'tag_reuters.com_2014_newsml_KBN0FL0NM:10'
        provider = get_resource_service('ingest_providers').find_one(name=provider_name, req=None)
        provider_service = self._get_provider_service(provider)
        provider_service.provider = provider
        provider_service.URL = provider.get('config', {}).get('url')
        items = provider_service.fetch_ingest(guid)

        for item in items:
            item['ingest_provider'] = provider['_id']
            item['expiry'] = utcnow() + timedelta(hours=11)

        # ingest the items
        self.ingest_items(items, provider, provider_service)

        items = provider_service.fetch_ingest(guid)
        for item in items:
            item['ingest_provider'] = provider['_id']
            item['expiry'] = utcnow() + timedelta(hours=11)

        # ingest them again
        self.ingest_items(items, provider, provider_service)

        # 12 files in grid fs
        current_files = self.app.media.fs('upload').find()
        self.assertEqual(12, current_files.count())
开发者ID:MiczFlor,项目名称:superdesk-core,代码行数:27,代码来源:update_ingest_tests.py


示例15: test_ingest_cancellation

 def test_ingest_cancellation(self):
     provider_name = 'reuters'
     guid = 'tag_reuters.com_2016_newsml_L1N14N0FF:978556838'
     provider = get_resource_service('ingest_providers').find_one(name=provider_name, req=None)
     provider_service = self._get_provider_service(provider)
     provider_service.provider = provider
     provider_service.URL = provider.get('config', {}).get('url')
     items = provider_service.fetch_ingest(guid)
     for item in items:
         item['ingest_provider'] = provider['_id']
         item['expiry'] = utcnow() + timedelta(hours=11)
     self.ingest_items(items, provider, provider_service)
     guid = 'tag_reuters.com_2016_newsml_L1N14N0FF:1542761538'
     items = provider_service.fetch_ingest(guid)
     for item in items:
         item['ingest_provider'] = provider['_id']
         item['expiry'] = utcnow() + timedelta(hours=11)
     self.ingest_items(items, provider, provider_service)
     ingest_service = get_resource_service('ingest')
     lookup = {'uri': items[0].get('uri')}
     family_members = ingest_service.get_from_mongo(req=None, lookup=lookup)
     self.assertEqual(family_members.count(), 2)
     for relative in family_members:
         self.assertEqual(relative['pubstatus'], 'canceled')
         self.assertEqual(relative['state'], 'killed')
开发者ID:MiczFlor,项目名称:superdesk-core,代码行数:25,代码来源:update_ingest_tests.py


示例16: test_ingest_associated_item_renditions

    def test_ingest_associated_item_renditions(self):
        provider = {'feeding_service': 'ninjs', '_id': self.providers['ninjs']}
        provider_service = FileFeedingService()
        item = {
            'guid': 'foo',
            'type': 'text',
            'versioncreated': utcnow(),
            'associations': {
                'featuremedia': {
                    'guid': 'bar',
                    'type': 'picture',
                    'versioncreated': utcnow(),
                    'renditions': {
                        'original': {
                            'href': 'https://farm8.staticflickr.com/7300/9203849352_297ea4207d_z_d.jpg',
                            'mimetype': 'image/jpeg',
                            'width': 640,
                            'height': 426,
                        }
                    }
                }
            }
        }

        # avoid transfer_renditions call which would store the picture locally
        # and it would fetch it using superdesk url which doesn't work in test
        with patch('superdesk.io.commands.update_ingest.transfer_renditions'):
            status, ids = ingest.ingest_item(item, provider, provider_service)

        self.assertTrue(status)
        self.assertEqual(2, len(ids))
        self.assertIn('thumbnail', item['associations']['featuremedia']['renditions'])
开发者ID:sjunaid,项目名称:superdesk-core,代码行数:32,代码来源:update_ingest_tests.py


示例17: test_bulletin_builder_formatter

    def test_bulletin_builder_formatter(self):
        article = {
            config.ID_FIELD: '123',
            config.VERSION: 2,
            'source': 'AAP',
            'anpa_category': [{'qcode': 'a'}],
            'headline': 'This is a test headline',
            'byline': 'joe',
            'slugline': 'slugline',
            'subject': [{'qcode': '02011001'}],
            'anpa_take_key': 'take_key',
            'unique_id': '1',
            'type': 'preformatted',
            'body_html': 'The story body',
            'abstract': 'abstract',
            'word_count': '1',
            'priority': '1',
            'firstcreated': utcnow(),
            'versioncreated': utcnow(),
            'lock_user': ObjectId()
        }

        subscriber = self.app.data.find('subscribers', None, None)[0]
        seq, item = self._formatter.format(article, subscriber)[0]
        item = json.loads(item)
        self.assertGreater(int(seq), 0)
        self.assertEqual(article[config.ID_FIELD], item.get('id'))
        self.assertEqual(article[config.VERSION], item.get('version'))
        self.assertEqual(article[ITEM_TYPE], item.get(ITEM_TYPE))
        self.assertEqual(article.get(PACKAGE_TYPE, ''), item.get(PACKAGE_TYPE))
        self.assertEqual(article['headline'], item.get('headline'))
        self.assertEqual(article['slugline'], item.get('slugline'))
        formatted_item = json.loads(item.get('data'))
        self.assertEqual(article['headline'], formatted_item['headline'])
开发者ID:akintolga,项目名称:superdesk-aap,代码行数:34,代码来源:aap_bulletinbuilder_formatter_tests.py


示例18: test_get_article_ids

 def test_get_article_ids(self):
     provider_name = 'reuters'
     provider, provider_service = self.setup_reuters_provider()
     ids = provider_service._get_article_ids('channel1', utcnow(), utcnow() + timedelta(minutes=-10))
     self.assertEqual(len(ids), 3)
     provider = get_resource_service('ingest_providers').find_one(name=provider_name, req=None)
     self.assertEqual(provider['tokens']['poll_tokens']['channel1'], 'ExwaY31kfnR2Z2J1cWZ2YnxoYH9kfw==')
开发者ID:liveblog,项目名称:superdesk-core,代码行数:7,代码来源:update_ingest_tests.py


示例19: transmit_subscriber_items

def transmit_subscriber_items(self, queue_items, subscriber):
    # Attempt to obtain a lock for transmissions to the subscriber
    lock_name = get_lock_id("Subscriber", "Transmit", subscriber)

    if not lock(lock_name, expire=610):
        return

    for queue_item in queue_items:
        publish_queue_service = get_resource_service(PUBLISH_QUEUE)
        log_msg = (
            "_id: {_id}  item_id: {item_id}  state: {state} "
            "item_version: {item_version} headline: {headline}".format(**queue_item)
        )
        try:
            # check the status of the queue item
            queue_item = publish_queue_service.find_one(req=None, _id=queue_item[config.ID_FIELD])
            if queue_item.get("state") not in [QueueState.PENDING.value, QueueState.RETRYING.value]:
                logger.info(
                    "Transmit State is not pending/retrying for queue item: {}. It is in {}".format(
                        queue_item.get(config.ID_FIELD), queue_item.get("state")
                    )
                )
                continue

            # update the status of the item to in-progress
            queue_update = {"state": "in-progress", "transmit_started_at": utcnow()}
            publish_queue_service.patch(queue_item.get(config.ID_FIELD), queue_update)
            logger.info("Transmitting queue item {}".format(log_msg))

            destination = queue_item["destination"]
            transmitter = superdesk.publish.registered_transmitters[destination.get("delivery_type")]
            transmitter.transmit(queue_item)
            logger.info("Transmitted queue item {}".format(log_msg))
        except Exception as e:
            logger.exception("Failed to transmit queue item {}".format(log_msg))

            max_retry_attempt = app.config.get("MAX_TRANSMIT_RETRY_ATTEMPT")
            retry_attempt_delay = app.config.get("TRANSMIT_RETRY_ATTEMPT_DELAY_MINUTES")
            try:
                orig_item = publish_queue_service.find_one(req=None, _id=queue_item["_id"])
                updates = {config.LAST_UPDATED: utcnow()}

                if orig_item.get("retry_attempt", 0) < max_retry_attempt and not isinstance(
                    e, PublishHTTPPushClientError
                ):

                    updates["retry_attempt"] = orig_item.get("retry_attempt", 0) + 1
                    updates["state"] = QueueState.RETRYING.value
                    updates["next_retry_attempt_at"] = utcnow() + timedelta(minutes=retry_attempt_delay)
                else:
                    # all retry attempts exhausted marking the item as failed.
                    updates["state"] = QueueState.FAILED.value

                publish_queue_service.system_update(orig_item.get(config.ID_FIELD), updates, orig_item)
            except:
                logger.error("Failed to set the state for failed publish queue item {}.".format(queue_item["_id"]))

    # Release the lock for the subscriber
    unlock(lock_name)
开发者ID:hlmnrmr,项目名称:superdesk-core,代码行数:59,代码来源:publish_content.py


示例20: update_times

    def update_times(self, doc):
        task = doc.get("task", {})
        status = task.get("status", None)
        if status == "in_progress":
            task.setdefault("started_at", utcnow())

        if status == "done":
            task.setdefault("finished_at", utcnow())
开发者ID:verifiedpixel,项目名称:superdesk,代码行数:8,代码来源:tasks.py



注:本文中的superdesk.utc.utcnow函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。


鲜花

握手

雷人

路过

鸡蛋
该文章已有0人参与评论

请发表评论

全部评论

专题导读
上一篇:
Python workflow.is_workflow_state_transition_valid函数代码示例发布时间:2022-05-27
下一篇:
Python utc.get_expiry_date函数代码示例发布时间:2022-05-27
热门推荐
阅读排行榜

扫描微信二维码

查看手机版网站

随时了解更新最新资讯

139-2527-9053

在线客服(服务时间 9:00~18:00)

在线QQ客服
地址:深圳市南山区西丽大学城创智工业园
电邮:jeky_zhao#qq.com
移动电话:139-2527-9053

Powered by 互联科技 X3.4© 2001-2213 极客世界.|Sitemap