callpoint.py 9.83 KB
Newer Older
Vangelis Koukis's avatar
Vangelis Koukis committed
1
# Copyright (C) 2010-2014 GRNET S.A.
Giorgos Korfiatis's avatar
Giorgos Korfiatis committed
2
#
Vangelis Koukis's avatar
Vangelis Koukis committed
3 4 5 6
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
Giorgos Korfiatis's avatar
Giorgos Korfiatis committed
7
#
Vangelis Koukis's avatar
Vangelis Koukis committed
8 9 10 11
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
Giorgos Korfiatis's avatar
Giorgos Korfiatis committed
12
#
Vangelis Koukis's avatar
Vangelis Koukis committed
13 14
# You should have received a copy of the GNU General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.
Giorgos Korfiatis's avatar
Giorgos Korfiatis committed
15

16
from datetime import datetime
17
from django.db.models import Q
18
from astakos.quotaholder_app.exception import (
19
    QuotaholderError,
20
    NoCommissionError,
21
    CorruptedError,
22
    NoHoldingError,
23
)
24

25
from astakos.quotaholder_app.commission import (
26
    Import, Release, Operations, finalize, undo)
Giorgos Korfiatis's avatar
Giorgos Korfiatis committed
27

Giorgos Korfiatis's avatar
Giorgos Korfiatis committed
28
from astakos.quotaholder_app.models import (
29
    Holding, Commission, Provision, ProvisionLog)
30

31

32 33 34 35
def format_datetime(d):
    return d.strftime('%Y-%m-%dT%H:%M:%S.%f')[:24]


36 37 38 39 40
def get_quota(holders=None, sources=None, resources=None, flt=None):
    if flt is None:
        flt = Q()

    holdings = Holding.objects.filter(flt)
41

42 43
    if holders is not None:
        holdings = holdings.filter(holder__in=holders)
44

45 46
    if sources is not None:
        holdings = holdings.filter(source__in=sources)
47

48 49
    if resources is not None:
        holdings = holdings.filter(resource__in=resources)
50

51 52 53
    quotas = {}
    for holding in holdings:
        key = (holding.holder, holding.source, holding.resource)
54
        value = (holding.limit, holding.usage_min, holding.usage_max)
55
        quotas[key] = value
56

57
    return quotas
58 59


60 61 62 63 64 65 66
def delete_quota(keys):
    for holder, source, resource in keys:
        Holding.objects.filter(holder=holder,
                               source=source,
                               resource=resource).delete()


67 68
def _get_holdings_for_update(holding_keys, resource=None, delete=False):
    flt = Q(resource=resource) if resource is not None else Q()
69
    holders = set(holder for (holder, source, resource) in holding_keys)
70
    objs = Holding.objects.filter(flt, holder__in=holders).order_by('pk')
71
    hs = objs.select_for_update()
72

73
    keys = set(holding_keys)
74
    holdings = {}
75
    put_back = []
76 77
    for h in hs:
        key = h.holder, h.source, h.resource
78 79 80 81 82 83 84 85
        if key in keys:
            holdings[key] = h
        else:
            put_back.append(h)

    if delete:
        objs.delete()
        Holding.objects.bulk_create(put_back)
86
    return holdings
87 88


89 90 91 92 93 94 95 96 97
def _mkProvision(key, quantity):
    holder, source, resource = key
    return {'holder': holder,
            'source': source,
            'resource': resource,
            'quantity': quantity,
            }


98
def set_quota(quotas, resource=None):
99
    holding_keys = [key for (key, limit) in quotas]
100 101
    holdings = _get_holdings_for_update(
        holding_keys, resource=resource, delete=True)
102

103
    new_holdings = {}
104
    for key, limit in quotas:
105 106 107
        holder, source, res = key
        if resource is not None and resource != res:
            continue
108 109
        h = Holding(holder=holder,
                    source=source,
110
                    resource=res,
111
                    limit=limit)
112
        try:
113 114 115
            h_old = holdings[key]
            h.usage_min = h_old.usage_min
            h.usage_max = h_old.usage_max
116
            h.id = h_old.id
117
        except KeyError:
118 119 120 121
            pass
        new_holdings[key] = h

    Holding.objects.bulk_create(new_holdings.values())
122

123

124 125 126 127 128 129 130 131
def _merge_same_keys(provisions):
    prov_dict = _partition_by(lambda t: t[0], provisions, lambda t: t[1])
    tuples = []
    for key, values in prov_dict.iteritems():
        tuples.append((key, sum(values)))
    return tuples


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

136
    provisions = _merge_same_keys(provisions)
137 138 139 140 141 142 143 144
    keys = [key for (key, value) in provisions]
    holdings = _get_holdings_for_update(keys)
    try:
        for key, quantity in provisions:
            # Target
            try:
                th = holdings[key]
            except KeyError:
145
                m = ("There is no such holding %s" % unicode(key))
146 147 148 149 150 151 152 153 154
                provision = _mkProvision(key, quantity)
                raise NoHoldingError(m,
                                     provision=provision)

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

            else:  # release
                abs_quantity = -quantity
155
                operations.prepare(Release, th, abs_quantity, False)
156 157 158 159

            holdings[key] = th
            provisions_to_create.append((key, quantity))
    except QuotaholderError:
160
        operations.revert()
161 162
        raise

163 164
    commission = Commission.objects.create(clientkey=clientkey,
                                           name=name,
165
                                           issue_datetime=datetime.now())
166
    ps = []
167
    for (holder, source, resource), quantity in provisions_to_create:
168 169 170 171 172 173
        ps.append(Provision(serial=commission,
                            holder=holder,
                            source=source,
                            resource=resource,
                            quantity=quantity))
    Provision.objects.bulk_create(ps)
174 175 176 177

    return commission.serial


178
def _log_provision(commission, provision, holding, log_datetime, reason):
179 180 181 182 183 184 185 186

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

195
    return ProvisionLog(**kwargs)
196 197 198 199 200 201 202 203 204 205 206 207


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

    commissions = {}
    for c in cs:
        commissions[c.serial] = c
    return commissions


208 209 210
def _partition_by(f, l, convert=None):
    if convert is None:
        convert = lambda x: x
211 212 213 214
    d = {}
    for x in l:
        group = f(x)
        group_l = d.get(group, [])
215
        group_l.append(convert(x))
216 217 218 219
        d[group] = group_l
    return d


220
def resolve_pending_commissions(clientkey, accept_set=None, reject_set=None,
221
                                reason=''):
222 223 224 225 226
    if accept_set is None:
        accept_set = []
    if reject_set is None:
        reject_set = []

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

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

244
    log_datetime = datetime.now()
245 246 247 248 249 250 251 252 253 254

    accepted, rejected, notFound = [], [], []
    for serial, accept in actions.iteritems():
        commission = commissions.get(serial)
        if commission is None:
            notFound.append(serial)
            continue

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

255
        ps = provisions.get(serial, [])
256 257
        provision_ids = []
        plog = []
258 259 260 261
        for pv in ps:
            key = pv.holding_key()
            h = holdings.get(key)
            if h is None:
262
                raise CorruptedError("Corrupted provision '%s'" % str(key))
Christos Stavrakakis's avatar
Christos Stavrakakis committed
263

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

            prefix = 'ACCEPT:' if accept else 'REJECT:'
            comm_reason = prefix + reason[-121:]
274 275 276 277
            plog.append(
                _log_provision(commission, pv, h, log_datetime, comm_reason))
        Provision.objects.filter(id__in=provision_ids).delete()
        ProvisionLog.objects.bulk_create(plog)
278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294
        commission.delete()
    return accepted, rejected, notFound, conflicting


def resolve_pending_commission(clientkey, serial, accept=True):
    if accept:
        ok, notOk, notF, confl = resolve_pending_commissions(
            clientkey=clientkey, accept_set=[serial])
    else:
        notOk, ok, notF, confl = resolve_pending_commissions(
            clientkey=clientkey, reject_set=[serial])

    assert notOk == confl == []
    assert ok + notF == [serial]
    return bool(ok)


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


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

308
    objs = Provision.objects
309 310 311 312 313 314
    provisions = objs.filter(serial=commission)

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

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