This repository has been archived by the owner on Dec 7, 2022. It is now read-only.
/
consumer.py
92 lines (83 loc) · 3.83 KB
/
consumer.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
# -*- coding: utf-8 -*-
#
# Copyright © 2012 Red Hat, Inc.
#
# This software is licensed to you under the GNU General Public License as
# published by the Free Software Foundation; either version 2 of the License
# (GPLv2) or (at your option) any later version.
# There is NO WARRANTY for this software, express or implied, including the
# implied warranties of MERCHANTABILITY, NON-INFRINGEMENT, or FITNESS FOR A
# PARTICULAR PURPOSE.
# You should have received a copy of GPLv2 along with this software; if not,
# see http://www.gnu.org/licenses/old-licenses/gpl-2.0.txt
"""
Itinerary creation for complex consumer operations.
"""
from pulp.common.tags import action_tag, resource_tag
from pulp.server import config as pulp_config
from pulp.server.dispatch import constants as dispatch_constants
from pulp.server.dispatch.call import CallRequest
from pulp.server.managers import factory as managers_factory
def consumer_content_install_itinerary(consumer_id, units, options):
"""
Create an itinerary for consumer content installation.
@param consumer_id: unique id of the consumer
@type consumer_id: str
@param units: units to install
@type units: list or tuple
@param options: options to pass to the install manager
@type options: dict or None
@return: list of call requests
@rtype: list
"""
manager = managers_factory.consumer_agent_manager()
args = [consumer_id]
kwargs = {'units': units, 'options': options}
weight = pulp_config.config.getint('tasks', 'consumer_content_weight')
tags = [resource_tag(dispatch_constants.RESOURCE_CONSUMER_TYPE, consumer_id),
action_tag('unit_install')]
call_request = CallRequest(manager.install_content, args, kwargs, weight=weight, tags=tags, archive=True, asynchronous=True)
call_request.reads_resource(dispatch_constants.RESOURCE_CONSUMER_TYPE, consumer_id)
return [call_request]
def consumer_content_update_itinerary(consumer_id, units, options):
"""
Create an itinerary for consumer content update.
@param consumer_id: unique id of the consumer
@type consumer_id: str
@param units: units to update
@type units: list or tuple
@param options: options to pass to the update manager
@type options: dict or None
@return: list of call requests
@rtype: list
"""
manager = managers_factory.consumer_agent_manager()
args = [consumer_id]
kwargs = {'units': units, 'options': options}
weight = pulp_config.config.getint('tasks', 'consumer_content_weight')
tags = [resource_tag(dispatch_constants.RESOURCE_CONSUMER_TYPE, consumer_id),
action_tag('unit_update')]
call_request = CallRequest(manager.update_content, args, kwargs, weight=weight, tags=tags, archive=True, asynchronous=True)
call_request.reads_resource(dispatch_constants.RESOURCE_CONSUMER_TYPE, consumer_id)
return [call_request]
def consumer_content_uninstall_itinerary(consumer_id, units, options):
"""
Create an itinerary for consumer content uninstall.
@param consumer_id: unique id of the consumer
@type consumer_id: str
@param units: units to uninstall
@type units: list or tuple
@param options: options to pass to the uninstall manager
@type options: dict or None
@return: list of call requests
@rtype: list
"""
manager = managers_factory.consumer_agent_manager()
args = [consumer_id]
kwargs = {'units': units, 'options': options}
weight = pulp_config.config.getint('tasks', 'consumer_content_weight')
tags = [resource_tag(dispatch_constants.RESOURCE_CONSUMER_TYPE, consumer_id),
action_tag('unit_uninstall')]
call_request = CallRequest(manager.uninstall_content, args, kwargs, weight=weight, tags=tags, archive=True, asynchronous=True)
call_request.reads_resource(dispatch_constants.RESOURCE_CONSUMER_TYPE, consumer_id)
return [call_request]