Passed
Push — master ( 1bb58d...3058b1 )
by Jochen
02:25
created

byceps.services.snippet.mountpoint_service   A

Complexity

Total Complexity 7

Size/Duplication

Total Lines 81
Duplicated Lines 0 %

Importance

Changes 0
Metric Value
eloc 43
dl 0
loc 81
rs 10
c 0
b 0
f 0
wmc 7

6 Functions

Rating   Name   Duplication   Size   Complexity  
A create_mountpoint() 0 10 1
A delete_mountpoint() 0 9 2
A find_mountpoint() 0 5 1
A find_current_snippet_version_for_mountpoint() 0 13 1
A get_mountpoints_for_site() 0 7 1
A _db_entity_to_mountpoint() 0 7 1
1
"""
2
byceps.services.snippet.mountpoint_service
3
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
4
5
:Copyright: 2006-2019 Jochen Kupperschmidt
6
:License: Modified BSD, see LICENSE for details.
7
"""
8
9
from typing import Optional, Set
10
11
from ...database import db
12
13
from ..site.transfer.models import SiteID
14
15
from .models.mountpoint import Mountpoint as DbMountpoint
16
from .models.snippet import CurrentVersionAssociation, Snippet, SnippetVersion
17
from .transfer.models import Mountpoint, MountpointID, SnippetID
18
19
20
def create_mountpoint(
21
    site_id: SiteID, endpoint_suffix: str, url_path: str, snippet_id: SnippetID
22
) -> Mountpoint:
23
    """Create a mountpoint."""
24
    mountpoint = DbMountpoint(site_id, endpoint_suffix, url_path, snippet_id)
25
26
    db.session.add(mountpoint)
27
    db.session.commit()
28
29
    return _db_entity_to_mountpoint(mountpoint)
30
31
32
def delete_mountpoint(mountpoint_id: MountpointID) -> None:
33
    """Delete the mountpoint."""
34
    mountpoint = DbMountpoint.query.get(mountpoint_id)
35
36
    if mountpoint is None:
37
        raise ValueError(f"Unknown mountpoint ID '{mountpoint_id}'.")
38
39
    db.session.delete(mountpoint)
40
    db.session.commit()
41
42
43
def find_mountpoint(mountpoint_id: MountpointID) -> Optional[Mountpoint]:
44
    """Return the mountpoint with that id, or `None` if not found."""
45
    mountpoint = DbMountpoint.query.get(mountpoint_id)
46
47
    return _db_entity_to_mountpoint(mountpoint)
48
49
50
def get_mountpoints_for_site(site_id: SiteID) -> Set[Mountpoint]:
51
    """Return all mountpoints for that site."""
52
    mountpoints = DbMountpoint.query \
53
        .filter_by(site_id=site_id) \
54
        .all()
55
56
    return {_db_entity_to_mountpoint(mp) for mp in mountpoints}
57
58
59
def find_current_snippet_version_for_mountpoint(
60
    site_id: SiteID, endpoint_suffix: str
61
) -> SnippetVersion:
62
    """Return the current version of the snippet mounted at that
63
    endpoint of that site, or `None` if not found.
64
    """
65
    return SnippetVersion.query \
66
        .join(CurrentVersionAssociation) \
67
        .join(Snippet) \
68
        .join(DbMountpoint) \
69
        .filter(DbMountpoint.site_id == site_id) \
70
        .filter(DbMountpoint.endpoint_suffix == endpoint_suffix) \
71
        .one_or_none()
72
73
74
def _db_entity_to_mountpoint(entity: DbMountpoint) -> Mountpoint:
75
    return Mountpoint(
76
        entity.id,
77
        entity.site_id,
78
        entity.endpoint_suffix,
79
        entity.url_path,
80
        entity.snippet_id,
81
    )
82