• Home
  • Features
  • Pricing
  • Docs
  • Announcements
  • Sign In

NaturalHistoryMuseum / data-importer / #12

07 Aug 2025 10:46AM UTC coverage: 79.133% (-0.1%) from 79.237%
#12

push

coveralls-python

web-flow
Merge dd79f4361 into edec251d1

196 of 294 new or added lines in 26 files covered. (66.67%)

5 existing lines in 2 files now uncovered.

1259 of 1591 relevant lines covered (79.13%)

0.79 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

0.0
/dataimporter/cli/utils.py
1
from functools import partial
×
2
from pathlib import Path
×
NEW
3
from typing import Any, Optional
×
4

NEW
5
import click
×
NEW
6
import psycopg
×
NEW
7
from click import Context, Parameter
×
8
from rich.console import Console
×
9

NEW
10
from dataimporter.lib.config import Config, ConfigLoadError, load
×
11

12
# environment variable name for config path setting
NEW
13
CONFIG_ENV_VAR = 'DIMP_CONFIG'
×
14

15
# global console for all to use
NEW
16
console: Console = Console(log_time_format='[%Y-%m-%d %H:%M:%S] ')
×
17

18

19
class ConfigType(click.Path):
×
20
    """
21
    Click type allowing CLI functions to get a config object from a path.
22
    """
23

NEW
24
    name = 'config'
×
25

26
    def __init__(self):
×
27
        super().__init__(
×
28
            exists=True, file_okay=True, dir_okay=False, readable=True, path_type=Path
29
        )
30

31
    def convert(
×
32
        self, value: Any, param: Optional[Parameter], ctx: Optional[Context]
33
    ) -> Config:
34
        """
35
        Convert the given value to a Config object.
36

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

46
        path: Path = Path(super().convert(value, param, ctx))
×
47
        try:
×
48
            return load(path)
×
49
        except ConfigLoadError as e:
×
50
            self.fail(
×
51
                f'Failed to load config from {path} due to {e.reason}',
52
                param,
53
                ctx,
54
            )
55
        except Exception as e:
×
56
            self.fail(
×
57
                f'Failed to load config from {path} due to {str(e)}',
58
                param,
59
                ctx,
60
            )
61

62

63
# decorator which adds the config click arg to any click command function
64
with_config = partial(
×
65
    click.argument, 'config', type=ConfigType(), envvar=CONFIG_ENV_VAR
66
)
67

68

69
def get_api_key(db_dsn: str, admin_user: str) -> Optional[str]:
×
70
    """
71
    Get the API key for the admin user and return it, or None if we can't get the key.
72

73
    :param db_dsn: the database datasource name to connect to
74
    :param admin_user: the name of the admin user to get the API key for
75
    :return: the API key
76
    """
77
    with psycopg.connect(db_dsn) as connection:
×
78
        with connection.cursor() as cursor:
×
79
            cursor.execute(
×
80
                'select apikey from public.user where name = %s;', (admin_user,)
81
            )
82
            row = cursor.fetchone()
×
83
            return None if row is None else row[0]
×
STATUS · Troubleshooting · Open an Issue · Sales · Support · CAREERS · ENTERPRISE · START FREE · SCHEDULE DEMO
ANNOUNCEMENTS · TWITTER · TOS & SLA · Supported CI Services · What's a CI service? · Automated Testing

© 2026 Coveralls, Inc