~homeworkprod/byceps

ref: 4237b3ec9496efe95dcce82bea3207ab9de4d520 byceps/byceps/services/shop/storefront/service.py -rw-r--r-- 3.9 KiB
4237b3ec — Jochen Kupperschmidt Move ticketing blueprint into `site` subpackage 2 years ago
                                                                                
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
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
"""
byceps.services.shop.storefront.service
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

:Copyright: 2006-2020 Jochen Kupperschmidt
:License: Modified BSD, see LICENSE for details.
"""

from typing import List, Optional, Set

from ....database import db

from ..catalog.transfer.models import CatalogID
from ..sequence.transfer.models import NumberSequenceID
from ..shop.transfer.models import ShopID

from .models import Storefront as DbStorefront
from .transfer.models import Storefront, StorefrontID


class UnknownStorefrontId(ValueError):
    pass


def create_storefront(
    storefront_id: StorefrontID,
    shop_id: ShopID,
    order_number_sequence_id: NumberSequenceID,
    closed: bool,
    *,
    catalog_id: Optional[CatalogID] = None,
) -> Storefront:
    """Create a storefront."""
    storefront = DbStorefront(
        storefront_id,
        shop_id,
        order_number_sequence_id,
        closed,
        catalog_id=catalog_id,
    )

    db.session.add(storefront)
    db.session.commit()

    return _db_entity_to_storefront(storefront)


def update_storefront(
    storefront_id: StorefrontID,
    catalog_id: CatalogID,
    order_number_sequence_id: NumberSequenceID,
    closed: bool,
) -> Storefront:
    """Update a storefront."""
    storefront = _get_db_storefront(storefront_id)

    storefront.catalog_id = catalog_id
    storefront.order_number_sequence_id = order_number_sequence_id
    storefront.closed = closed

    db.session.commit()

    return _db_entity_to_storefront(storefront)


def delete_storefront(storefront_id: StorefrontID) -> None:
    """Delete a storefront."""
    db.session.query(DbStorefront) \
        .filter_by(id=storefront_id) \
        .delete()

    db.session.commit()


def find_storefront(storefront_id: StorefrontID) -> Optional[Storefront]:
    """Return the storefront with that id, or `None` if not found."""
    storefront = _find_db_storefront(storefront_id)

    if storefront is None:
        return None

    return _db_entity_to_storefront(storefront)


def _find_db_storefront(storefront_id: StorefrontID) -> Optional[DbStorefront]:
    """Return the database entity for the storefront with that id, or `None`
    if not found.
    """
    return DbStorefront.query.get(storefront_id)


def get_storefront(storefront_id: StorefrontID) -> Storefront:
    """Return the storefront with that id, or raise an exception."""
    storefront = find_storefront(storefront_id)

    if storefront is None:
        raise UnknownStorefrontId(storefront_id)

    return storefront


def _get_db_storefront(storefront_id: StorefrontID) -> DbStorefront:
    """Return the database entity for the storefront with that id.

    Raise an exception if not found.
    """
    storefront = _find_db_storefront(storefront_id)

    if storefront is None:
        raise UnknownStorefrontId(storefront_id)

    return storefront


def find_storefronts(storefront_ids: Set[StorefrontID]) -> List[Storefront]:
    """Return the storefronts with those IDs."""
    if not storefront_ids:
        return []

    storefronts = DbStorefront.query \
        .filter(DbStorefront.id.in_(storefront_ids)) \
        .all()

    return [_db_entity_to_storefront(storefront) for storefront in storefronts]


def get_all_storefronts() -> List[Storefront]:
    """Return all storefronts."""
    storefronts = DbStorefront.query.all()

    return [_db_entity_to_storefront(storefront) for storefront in storefronts]


def get_storefronts_for_shop(shop_id: ShopID) -> Set[Storefront]:
    """Return all storefronts for that shop."""
    rows = DbStorefront.query \
        .filter_by(shop_id=shop_id) \
        .all()

    return {_db_entity_to_storefront(row) for row in rows}


def _db_entity_to_storefront(storefront: DbStorefront) -> Storefront:
    return Storefront(
        storefront.id,
        storefront.shop_id,
        storefront.catalog_id,
        storefront.order_number_sequence_id,
        storefront.closed,
    )