Skip to content

Utils

ConfigType

Bases: Path

Click type allowing CLI functions to get a config object from a path.

Source code in dataimporter/cli/utils.py
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
class ConfigType(click.Path):
    """
    Click type allowing CLI functions to get a config object from a path.
    """

    name = 'config'

    def __init__(self):
        super().__init__(
            exists=True, file_okay=True, dir_okay=False, readable=True, path_type=Path
        )

    def convert(
        self, value: Any, param: Optional[Parameter], ctx: Optional[Context]
    ) -> Config:
        """
        Convert the given value to a Config object.

        :param value: the value passed from Click, hopefully this is a path of some kind
        :param param: the parameter that is using this type to convert its value. May be
            None.
        :param ctx: the current context that arrived at this value. May be None.
        :return: a config object
        """
        if isinstance(value, Config):
            return value

        path: Path = Path(super().convert(value, param, ctx))
        try:
            return load(path)
        except ConfigLoadError as e:
            self.fail(
                f'Failed to load config from {path} due to {e.reason}',
                param,
                ctx,
            )
        except Exception as e:
            self.fail(
                f'Failed to load config from {path} due to {str(e)}',
                param,
                ctx,
            )

convert(value, param, ctx)

Convert the given value to a Config object.

Parameters:

Name Type Description Default
value Any

the value passed from Click, hopefully this is a path of some kind

required
param Optional[Parameter]

the parameter that is using this type to convert its value. May be None.

required
ctx Optional[Context]

the current context that arrived at this value. May be None.

required

Returns:

Type Description
Config

a config object

Source code in dataimporter/cli/utils.py
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
def convert(
    self, value: Any, param: Optional[Parameter], ctx: Optional[Context]
) -> Config:
    """
    Convert the given value to a Config object.

    :param value: the value passed from Click, hopefully this is a path of some kind
    :param param: the parameter that is using this type to convert its value. May be
        None.
    :param ctx: the current context that arrived at this value. May be None.
    :return: a config object
    """
    if isinstance(value, Config):
        return value

    path: Path = Path(super().convert(value, param, ctx))
    try:
        return load(path)
    except ConfigLoadError as e:
        self.fail(
            f'Failed to load config from {path} due to {e.reason}',
            param,
            ctx,
        )
    except Exception as e:
        self.fail(
            f'Failed to load config from {path} due to {str(e)}',
            param,
            ctx,
        )

get_api_key(db_dsn, admin_user)

Get the API key for the admin user and return it, or None if we can't get the key.

Parameters:

Name Type Description Default
db_dsn str

the database datasource name to connect to

required
admin_user str

the name of the admin user to get the API key for

required

Returns:

Type Description
Optional[str]

the API key

Source code in dataimporter/cli/utils.py
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
def get_api_key(db_dsn: str, admin_user: str) -> Optional[str]:
    """
    Get the API key for the admin user and return it, or None if we can't get the key.

    :param db_dsn: the database datasource name to connect to
    :param admin_user: the name of the admin user to get the API key for
    :return: the API key
    """
    with psycopg.connect(db_dsn) as connection:
        with connection.cursor() as cursor:
            cursor.execute(
                'select apikey from public.user where name = %s;', (admin_user,)
            )
            row = cursor.fetchone()
            return None if row is None else row[0]