Statistics
| Branch: | Tag: | Revision:

root / snf-astakos-app / astakos / quotaholder_app / callpoint.py @ 5a0f9d6c

History | View | Annotate | Download (10.1 kB)

1
# Copyright 2012, 2013 GRNET S.A. All rights reserved.
2
#
3
# Redistribution and use in source and binary forms, with or
4
# without modification, are permitted provided that the following
5
# conditions are met:
6
#
7
#   1. Redistributions of source code must retain the above
8
#      copyright notice, this list of conditions and the following
9
#      disclaimer.
10
#
11
#   2. Redistributions in binary form must reproduce the above
12
#      copyright notice, this list of conditions and the following
13
#      disclaimer in the documentation and/or other materials
14
#      provided with the distribution.
15
#
16
# THIS SOFTWARE IS PROVIDED BY GRNET S.A. ``AS IS'' AND ANY EXPRESS
17
# OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
18
# WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
19
# PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL GRNET S.A OR
20
# CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
21
# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
22
# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF
23
# USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED
24
# AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
25
# LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN
26
# ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
27
# POSSIBILITY OF SUCH DAMAGE.
28
#
29
# The views and conclusions contained in the software and
30
# documentation are those of the authors and should not be
31
# interpreted as representing official policies, either expressed
32
# or implied, of GRNET S.A.
33

    
34
from datetime import datetime
35
from django.db.models import Q
36
from astakos.quotaholder_app.exception import (
37
    QuotaholderError,
38
    NoCommissionError,
39
    CorruptedError, InvalidDataError,
40
    NoHoldingError,
41
    DuplicateError)
42

    
43
from astakos.quotaholder_app.commission import (
44
    Import, Release, Operations, finalize, undo)
45

    
46
from astakos.quotaholder_app.models import (
47
    Holding, Commission, Provision, ProvisionLog)
48

    
49

    
50
def format_datetime(d):
51
    return d.strftime('%Y-%m-%dT%H:%M:%S.%f')[:24]
52

    
53

    
54
def get_quota(holders=None, sources=None, resources=None, flt=None):
55
    if flt is None:
56
        flt = Q()
57

    
58
    holdings = Holding.objects.filter(flt)
59

    
60
    if holders is not None:
61
        holdings = holdings.filter(holder__in=holders)
62

    
63
    if sources is not None:
64
        holdings = holdings.filter(source__in=sources)
65

    
66
    if resources is not None:
67
        holdings = holdings.filter(resource__in=resources)
68

    
69
    quotas = {}
70
    for holding in holdings:
71
        key = (holding.holder, holding.source, holding.resource)
72
        value = (holding.limit, holding.usage_min, holding.usage_max)
73
        quotas[key] = value
74

    
75
    return quotas
76

    
77

    
78
def _get_holdings_for_update(holding_keys, delete=False):
79
    holders = set(holder for (holder, source, resource) in holding_keys)
80
    objs = Holding.objects.filter(holder__in=holders).order_by('pk')
81
    hs = objs.select_for_update()
82

    
83
    keys = set(holding_keys)
84
    holdings = {}
85
    put_back = []
86
    for h in hs:
87
        key = h.holder, h.source, h.resource
88
        if key in keys:
89
            holdings[key] = h
90
        else:
91
            put_back.append(h)
92

    
93
    if delete:
94
        objs.delete()
95
        Holding.objects.bulk_create(put_back)
96
    return holdings
97

    
98

    
99
def _mkProvision(key, quantity):
100
    holder, source, resource = key
101
    return {'holder': holder,
102
            'source': source,
103
            'resource': resource,
104
            'quantity': quantity,
105
            }
106

    
107

    
108
def set_quota(quotas):
109
    holding_keys = [key for (key, limit) in quotas]
110
    holdings = _get_holdings_for_update(holding_keys, delete=True)
111

    
112
    new_holdings = {}
113
    for key, limit in quotas:
114
        holder, source, resource = key
115
        h = Holding(holder=holder,
116
                    source=source,
117
                    resource=resource,
118
                    limit=limit)
119
        try:
120
            h_old = holdings[key]
121
            h.usage_min = h_old.usage_min
122
            h.usage_max = h_old.usage_max
123
        except KeyError:
124
            pass
125
        new_holdings[key] = h
126

    
127
    Holding.objects.bulk_create(new_holdings.values())
128

    
129

    
130
def issue_commission(clientkey, provisions, name="", force=False):
131
    operations = Operations()
132
    provisions_to_create = []
133

    
134
    keys = [key for (key, value) in provisions]
135
    holdings = _get_holdings_for_update(keys)
136
    try:
137
        checked = []
138
        for key, quantity in provisions:
139
            if not isinstance(quantity, (int, long)):
140
                raise InvalidDataError("Malformed provision")
141

    
142
            if key in checked:
143
                m = "Duplicate provision for %s" % str(key)
144
                provision = _mkProvision(key, quantity)
145
                raise DuplicateError(m,
146
                                     provision=provision)
147
            checked.append(key)
148

    
149
            # Target
150
            try:
151
                th = holdings[key]
152
            except KeyError:
153
                m = ("There is no such holding %s" % str(key))
154
                provision = _mkProvision(key, quantity)
155
                raise NoHoldingError(m,
156
                                     provision=provision)
157

    
158
            if quantity >= 0:
159
                operations.prepare(Import, th, quantity, force)
160

    
161
            else:  # release
162
                abs_quantity = -quantity
163
                operations.prepare(Release, th, abs_quantity, False)
164

    
165
            holdings[key] = th
166
            provisions_to_create.append((key, quantity))
167

    
168
    except QuotaholderError:
169
        operations.revert()
170
        raise
171

    
172
    commission = Commission.objects.create(clientkey=clientkey,
173
                                           name=name,
174
                                           issue_datetime=datetime.now())
175
    for (holder, source, resource), quantity in provisions_to_create:
176
        Provision.objects.create(serial=commission,
177
                                 holder=holder,
178
                                 source=source,
179
                                 resource=resource,
180
                                 quantity=quantity)
181

    
182
    return commission.serial
183

    
184

    
185
def _log_provision(commission, provision, holding, log_datetime, reason):
186

    
187
    kwargs = {
188
        'serial':              commission.serial,
189
        'name':                commission.name,
190
        'holder':              holding.holder,
191
        'source':              holding.source,
192
        'resource':            holding.resource,
193
        'limit':               holding.limit,
194
        'usage_min':           holding.usage_min,
195
        'usage_max':           holding.usage_max,
196
        'delta_quantity':      provision.quantity,
197
        'issue_time':          format_datetime(commission.issue_datetime),
198
        'log_time':            format_datetime(log_datetime),
199
        'reason':              reason,
200
    }
201

    
202
    ProvisionLog.objects.create(**kwargs)
203

    
204

    
205
def _get_commissions_for_update(clientkey, serials):
206
    cs = Commission.objects.filter(
207
        clientkey=clientkey, serial__in=serials).select_for_update()
208

    
209
    commissions = {}
210
    for c in cs:
211
        commissions[c.serial] = c
212
    return commissions
213

    
214

    
215
def _partition_by(f, l):
216
    d = {}
217
    for x in l:
218
        group = f(x)
219
        group_l = d.get(group, [])
220
        group_l.append(x)
221
        d[group] = group_l
222
    return d
223

    
224

    
225
def resolve_pending_commissions(clientkey, accept_set=None, reject_set=None,
226
                                reason=''):
227
    if accept_set is None:
228
        accept_set = []
229
    if reject_set is None:
230
        reject_set = []
231

    
232
    actions = dict.fromkeys(accept_set, True)
233
    conflicting = set()
234
    for serial in reject_set:
235
        if actions.get(serial) is True:
236
            actions.pop(serial)
237
            conflicting.add(serial)
238
        else:
239
            actions[serial] = False
240

    
241
    conflicting = list(conflicting)
242
    serials = actions.keys()
243
    commissions = _get_commissions_for_update(clientkey, serials)
244
    ps = Provision.objects.filter(serial__in=serials).select_for_update()
245
    holding_keys = sorted(p.holding_key() for p in ps)
246
    holdings = _get_holdings_for_update(holding_keys)
247
    provisions = _partition_by(lambda p: p.serial_id, ps)
248

    
249
    log_datetime = datetime.now()
250

    
251
    accepted, rejected, notFound = [], [], []
252
    for serial, accept in actions.iteritems():
253
        commission = commissions.get(serial)
254
        if commission is None:
255
            notFound.append(serial)
256
            continue
257

    
258
        accepted.append(serial) if accept else rejected.append(serial)
259

    
260
        ps = provisions.get(serial, [])
261
        for pv in ps:
262
            key = pv.holding_key()
263
            h = holdings.get(key)
264
            if h is None:
265
                raise CorruptedError("Corrupted provision")
266

    
267
            quantity = pv.quantity
268
            action = finalize if accept else undo
269
            if quantity >= 0:
270
                action(Import, h, quantity)
271
            else:  # release
272
                action(Release, h, -quantity)
273

    
274
            prefix = 'ACCEPT:' if accept else 'REJECT:'
275
            comm_reason = prefix + reason[-121:]
276
            _log_provision(commission, pv, h, log_datetime, comm_reason)
277
            pv.delete()
278
        commission.delete()
279
    return accepted, rejected, notFound, conflicting
280

    
281

    
282
def resolve_pending_commission(clientkey, serial, accept=True):
283
    if accept:
284
        ok, notOk, notF, confl = resolve_pending_commissions(
285
            clientkey=clientkey, accept_set=[serial])
286
    else:
287
        notOk, ok, notF, confl = resolve_pending_commissions(
288
            clientkey=clientkey, reject_set=[serial])
289

    
290
    assert notOk == confl == []
291
    assert ok + notF == [serial]
292
    return bool(ok)
293

    
294

    
295
def get_pending_commissions(clientkey):
296
    pending = Commission.objects.filter(clientkey=clientkey)
297
    pending_list = pending.values_list('serial', flat=True)
298
    return list(pending_list)
299

    
300

    
301
def get_commission(clientkey, serial):
302
    try:
303
        commission = Commission.objects.get(clientkey=clientkey,
304
                                            serial=serial)
305
    except Commission.DoesNotExist:
306
        raise NoCommissionError(serial)
307

    
308
    objs = Provision.objects
309
    provisions = objs.filter(serial=commission)
310

    
311
    ps = [p.todict() for p in provisions]
312

    
313
    response = {'serial':     serial,
314
                'provisions': ps,
315
                'issue_time': commission.issue_datetime,
316
                'name':       commission.name,
317
                }
318
    return response