File: responses.py

package info (click to toggle)
python-moto 5.1.18-3
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid
  • size: 116,520 kB
  • sloc: python: 636,725; javascript: 181; makefile: 39; sh: 3
file content (60 lines) | stat: -rw-r--r-- 1,934 bytes parent folder | download | duplicates (2)
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
import json
import re

from moto.core.responses import BaseResponse

from .exceptions import InvalidRepositoryNameException
from .models import CodeCommitBackend, codecommit_backends


def _is_repository_name_valid(repository_name: str) -> bool:
    name_regex = re.compile(r"[\w\.-]+")
    result = name_regex.split(repository_name)
    if len(result) > 0:
        for match in result:
            if len(match) > 0:
                return False
    return True


class CodeCommitResponse(BaseResponse):
    def __init__(self) -> None:
        super().__init__(service_name="codecommit")

    @property
    def codecommit_backend(self) -> CodeCommitBackend:
        return codecommit_backends[self.current_account][self.region]

    def create_repository(self) -> str:
        if not _is_repository_name_valid(self._get_param("repositoryName")):
            raise InvalidRepositoryNameException()

        repository_metadata = self.codecommit_backend.create_repository(
            self._get_param("repositoryName"),
            self._get_param("repositoryDescription"),
        )

        return json.dumps({"repositoryMetadata": repository_metadata})

    def get_repository(self) -> str:
        if not _is_repository_name_valid(self._get_param("repositoryName")):
            raise InvalidRepositoryNameException()

        repository_metadata = self.codecommit_backend.get_repository(
            self._get_param("repositoryName")
        )

        return json.dumps({"repositoryMetadata": repository_metadata})

    def delete_repository(self) -> str:
        if not _is_repository_name_valid(self._get_param("repositoryName")):
            raise InvalidRepositoryNameException()

        repository_id = self.codecommit_backend.delete_repository(
            self._get_param("repositoryName")
        )

        if repository_id:
            return json.dumps({"repositoryId": repository_id})

        return json.dumps({})