~homeworkprod/byceps

ref: 4237b3ec9496efe95dcce82bea3207ab9de4d520 byceps/byceps/services/shop/catalog/service.py -rw-r--r-- 3.4 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
"""
byceps.services.shop.catalog.service
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

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

from typing import List, Optional

from ....database import db

from ..article.transfer.models import ArticleNumber

from .models import (
    Catalog as DbCatalog,
    CatalogArticle as DbCatalogArticle,
    Collection as DbCollection,
)
from .transfer.models import (
    Catalog,
    CatalogArticleID,
    CatalogID,
    Collection,
    CollectionID,
)


# catalog


def create_catalog(catalog_id: CatalogID, title: str) -> Catalog:
    """Create a catalog."""
    catalog = DbCatalog(catalog_id, title)

    db.session.add(catalog)
    db.session.commit()

    return _db_entity_to_catalog(catalog)


def find_catalog(catalog_id: CatalogID) -> Optional[Catalog]:
    """Return the catalog with that ID, or `None` if not found."""
    catalog = _find_db_catalog(catalog_id)

    if catalog is None:
        return None

    return _db_entity_to_catalog(catalog)


def _find_db_catalog(catalog_id: CatalogID) -> Optional[DbCatalog]:
    """Return the catalog database entity with that ID, or `None` if not
    found.
    """
    return DbCatalog.query.get(catalog_id)


def get_all_catalogs() -> List[Catalog]:
    """Return all catalogs."""
    catalogs = DbCatalog.query.all()

    return [_db_entity_to_catalog(catalog) for catalog in catalogs]


def _db_entity_to_catalog(catalog: DbCatalog) -> Catalog:
    return Catalog(
        catalog.id,
        catalog.title,
    )


# collection


def create_collection(catalog_id: CatalogID, title: str) -> Collection:
    """Create a collection."""
    catalog = _find_db_catalog(catalog_id)
    if catalog is None:
        raise ValueError(f'Unknown catalog ID "{catalog_id}"')

    collection = DbCollection(catalog_id, title)

    catalog.collections.append(collection)
    db.session.commit()

    return _db_entity_to_collection(collection)


def delete_collection(collection_id: CollectionID) -> None:
    """Delete the collection."""
    db.session.query(DbCollection) \
        .filter_by(id=collection_id) \
        .delete()

    db.session.commit()


def get_collections_for_catalog(catalog_id: CatalogID) -> List[Collection]:
    """Return the catalog's collections."""
    collections = DbCollection.query \
        .filter_by(catalog_id=catalog_id) \
        .order_by(DbCollection.position) \
        .all()

    return [_db_entity_to_collection(collection) for collection in collections]


def _db_entity_to_collection(collection: DbCollection) -> Collection:
    return Collection(
        collection.id,
        collection.catalog_id,
        collection.title,
        collection.position,
        [],
    )


# article assignment


def add_article_to_collection(
    article_number: ArticleNumber, collection_id: CollectionID
) -> CatalogArticleID:
    """Add article to collection."""
    collection = DbCollection.query.get(collection_id)
    if collection is None:
        raise ValueError(f'Unknown collection ID "{collection_id}"')

    catalog_article = DbCatalogArticle(collection_id, article_number)

    collection.catalog_articles.append(catalog_article)
    db.session.commit()

    return catalog_article.id


def remove_article_from_collection(
    catalog_article_id: CatalogArticleID,
) -> None:
    """Remove article from collection."""
    db.session.query(DbCatalogArticle) \
        .filter_by(id=catalog_article_id) \
        .delete()

    db.session.commit()