byceps.services.shop.catalog.service._find_db_collection() - Code Metrics - Inspection of "Extend shop catalog service" - byceps/byceps - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — main ( 2312e0...0e5c14 )

by Jochen

created 2022-03-22 02:14 UTC

_find_db_collection() A

↳ Parent: byceps.services.shop.catalog.service

Complexity

Conditions

Size

Total Lines	5
Code Lines	2

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	1
CRAP Score	1.125

Importance

Changes

Metric	Value
cc	1
eloc	2
nop	1
dl	0
loc	5
ccs	1
cts	2
cp	0.5
crap	1.125
rs	10
c	0
b	0
f	0

"""
byceps.services.shop.catalog.service
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

:Copyright: 2014-2022 Jochen Kupperschmidt
:License: Revised BSD (see `LICENSE` file for details)
"""

from __future__ import annotations
from typing import Optional

from sqlalchemy import delete, select

from ....database import db

from ..article.transfer.models import ArticleNumber
from ..shop.transfer.models import ShopID

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


# catalog


def create_catalog(shop_id: ShopID, title: str) -> Catalog:
    """Create a catalog."""
    db_catalog = DbCatalog(shop_id, title)

    db.session.add(db_catalog)
    db.session.commit()

    return _db_entity_to_catalog(db_catalog)


def update_catalog(catalog_id: CatalogID, title: str) -> Catalog:
    """Update a catalog."""
    db_catalog = _find_db_catalog(catalog_id)

    if db_catalog is None:
        raise ValueError(f'Unknown shop catalog ID "{catalog_id}"')

    db_catalog.title = title

    db.session.commit()

    return _db_entity_to_catalog(db_catalog)


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

    if db_catalog is None:
        return None

    return _db_entity_to_catalog(db_catalog)


def get_catalog(catalog_id: CatalogID) -> Catalog:
    """Return the catalog with that ID."""
    catalog = find_catalog(catalog_id)

    if catalog is None:
        raise ValueError(f'Unknown shop catalog ID "{catalog_id}"')

    return catalog


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


def get_catalogs_for_shop(shop_id: ShopID) -> list[Catalog]:
    """Return all catalogs for that shop."""
    db_catalogs = db.session.execute(
        select(DbCatalog)
        .filter_by(shop_id=shop_id)
    ).scalars().all()

    return [_db_entity_to_catalog(db_catalog) for db_catalog in db_catalogs]


def _db_entity_to_catalog(db_catalog: DbCatalog) -> Catalog:
    return Catalog(
        id=db_catalog.id,
        shop_id=db_catalog.shop_id,
        title=db_catalog.title,
    )


# collection


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

    db_collection = DbCollection(catalog_id, title)

    db_catalog.collections.append(db_collection)
    db.session.commit()

    return _db_entity_to_collection(db_collection)


def update_collection(collection_id: CollectionID, title: str) -> Collection:
    """Update a collection."""
    db_collection = _find_db_collection(collection_id)

    if db_collection is None:
        raise ValueError(f'Unknown shop collection ID "{collection_id}"')

    db_collection.title = title

    db.session.commit()

    return _db_entity_to_collection(db_collection)


def delete_collection(collection_id: CollectionID) -> None:
    """Delete the collection."""
    db.session.execute(
        delete(DbCollection)
        .where(DbCollection.id == collection_id)
    )
    db.session.commit()


def find_collection(collection_id: CollectionID) -> Optional[Collection]:
    """Return the collection with that ID, or `None` if not found."""
    db_collection = _find_db_collection(collection_id)

    if db_collection is None:
        return None

    return _db_entity_to_collection(db_collection)


def _find_db_collection(collection_id: CollectionID) -> Optional[DbCollection]:
    """Return the collection database entity with that ID, or `None` if
    not found.
    """
    return db.session.get(DbCollection, collection_id)


def get_collections_for_catalog(catalog_id: CatalogID) -> list[Collection]:
    """Return the catalog's collections."""
    db_collections = db.session.execute(
        select(DbCollection)
        .filter_by(catalog_id=catalog_id)
        .order_by(DbCollection.position)
    ).scalars().all()

    return [
        _db_entity_to_collection(db_collection)
        for db_collection in db_collections
    ]


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


# article assignment


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

    db_catalog_article = DbCatalogArticle(collection_id, article_number)

    db_collection.catalog_articles.append(db_catalog_article)
    db.session.commit()

    return db_catalog_article.id


def remove_article_from_collection(
    catalog_article_id: CatalogArticleID,
) -> None:
    """Remove article from collection."""
    db.session.execute(
        delete(DbCatalogArticle)
        .where(DbCatalogArticle.id == catalog_article_id)
    )
    db.session.commit()


1		"""
2		byceps.services.shop.catalog.service
3		~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
4
5		:Copyright: 2014-2022 Jochen Kupperschmidt
6		:License: Revised BSD (see `LICENSE` file for details)
7		"""
8
9	1	from __future__ import annotations
10	1	from typing import Optional
11
12	1	from sqlalchemy import delete, select
13
14	1	from ....database import db
15
16	1	from ..article.transfer.models import ArticleNumber
17	1	from ..shop.transfer.models import ShopID
18
19	1	from .dbmodels import (
20		Catalog as DbCatalog,
21		CatalogArticle as DbCatalogArticle,
22		Collection as DbCollection,
23		)
24	1	from .transfer.models import (
25		Catalog,
26		CatalogArticleID,
27		CatalogID,
28		Collection,
29		CollectionID,
30		)
31
32
33		# catalog
34
35
36	1	def create_catalog(shop_id: ShopID, title: str) -> Catalog:
37		"""Create a catalog."""
38		db_catalog = DbCatalog(shop_id, title)
39
40		db.session.add(db_catalog)
41		db.session.commit()
42
43		return _db_entity_to_catalog(db_catalog)
44
45
46	1	def update_catalog(catalog_id: CatalogID, title: str) -> Catalog:
47		"""Update a catalog."""
48		db_catalog = _find_db_catalog(catalog_id)
49
50		if db_catalog is None:
51		raise ValueError(f'Unknown shop catalog ID "{catalog_id}"')
52
53		db_catalog.title = title
54
55		db.session.commit()
56
57		return _db_entity_to_catalog(db_catalog)
58
59
60	1	def find_catalog(catalog_id: CatalogID) -> Optional[Catalog]:
61		"""Return the catalog with that ID, or `None` if not found."""
62		db_catalog = _find_db_catalog(catalog_id)
63
64		if db_catalog is None:
65		return None
66
67		return _db_entity_to_catalog(db_catalog)
68
69
70	1	def get_catalog(catalog_id: CatalogID) -> Catalog:
71		"""Return the catalog with that ID."""
72		catalog = find_catalog(catalog_id)
73
74		if catalog is None:
75		raise ValueError(f'Unknown shop catalog ID "{catalog_id}"')
76
77		return catalog
78
79
80	1	def _find_db_catalog(catalog_id: CatalogID) -> Optional[DbCatalog]:
81		"""Return the catalog database entity with that ID, or `None` if not
82		found.
83		"""
84		return db.session.get(DbCatalog, catalog_id)
85
86
87	1	def get_catalogs_for_shop(shop_id: ShopID) -> list[Catalog]:
88		"""Return all catalogs for that shop."""
89		db_catalogs = db.session.execute(
90		select(DbCatalog)
91		.filter_by(shop_id=shop_id)
92		).scalars().all()
93
94		return [_db_entity_to_catalog(db_catalog) for db_catalog in db_catalogs]
95
96
97	1	def _db_entity_to_catalog(db_catalog: DbCatalog) -> Catalog:
98		return Catalog(
99		id=db_catalog.id,
100		shop_id=db_catalog.shop_id,
101		title=db_catalog.title,
102		)
103
104
105		# collection
106
107
108	1	def create_collection(catalog_id: CatalogID, title: str) -> Collection:
109		"""Create a collection."""
110		db_catalog = _find_db_catalog(catalog_id)
111		if db_catalog is None:
112		raise ValueError(f'Unknown catalog ID "{catalog_id}"')
113
114		db_collection = DbCollection(catalog_id, title)
115
116		db_catalog.collections.append(db_collection)
117		db.session.commit()
118
119		return _db_entity_to_collection(db_collection)
120
121
122	1	def update_collection(collection_id: CollectionID, title: str) -> Collection:
123		"""Update a collection."""
124		db_collection = _find_db_collection(collection_id)
125
126		if db_collection is None:
127		raise ValueError(f'Unknown shop collection ID "{collection_id}"')
128
129		db_collection.title = title
130
131		db.session.commit()
132
133		return _db_entity_to_collection(db_collection)
134
135
136	1	def delete_collection(collection_id: CollectionID) -> None:
137		"""Delete the collection."""
138		db.session.execute(
139		delete(DbCollection)
140		.where(DbCollection.id == collection_id)
141		)
142		db.session.commit()
143
144
145	1	def find_collection(collection_id: CollectionID) -> Optional[Collection]:
146		"""Return the collection with that ID, or `None` if not found."""
147		db_collection = _find_db_collection(collection_id)
148
149		if db_collection is None:
150		return None
151
152		return _db_entity_to_collection(db_collection)
153
154
155	1	def _find_db_collection(collection_id: CollectionID) -> Optional[DbCollection]:
156		"""Return the collection database entity with that ID, or `None` if
157		not found.
158		"""
159		return db.session.get(DbCollection, collection_id)
160
161
162	1	def get_collections_for_catalog(catalog_id: CatalogID) -> list[Collection]:
163		"""Return the catalog's collections."""
164		db_collections = db.session.execute(
165		select(DbCollection)
166		.filter_by(catalog_id=catalog_id)
167		.order_by(DbCollection.position)
168		).scalars().all()
169
170		return [
171		_db_entity_to_collection(db_collection)
172		for db_collection in db_collections
173		]
174
175
176	1	def _db_entity_to_collection(db_collection: DbCollection) -> Collection:
177		return Collection(
178		id=db_collection.id,
179		catalog_id=db_collection.catalog_id,
180		title=db_collection.title,
181		position=db_collection.position,
182		article_numbers=[],
183		)
184
185
186		# article assignment
187
188
189	1	def add_article_to_collection(
190		article_number: ArticleNumber, collection_id: CollectionID
191		) -> CatalogArticleID:
192		"""Add article to collection."""
193		db_collection = db.session.get(DbCollection, collection_id)
194		if db_collection is None:
195		raise ValueError(f'Unknown collection ID "{collection_id}"')
196
197		db_catalog_article = DbCatalogArticle(collection_id, article_number)
198
199		db_collection.catalog_articles.append(db_catalog_article)
200		db.session.commit()
201
202		return db_catalog_article.id
203
204
205	1	def remove_article_from_collection(
206		catalog_article_id: CatalogArticleID,
207		) -> None:
208		"""Remove article from collection."""
209		db.session.execute(
210		delete(DbCatalogArticle)
211		.where(DbCatalogArticle.id == catalog_article_id)
212		)
213		db.session.commit()
214

byceps / byceps

Push — main ( 2312e0...0e5c14 )

_find_db_collection() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like