Skip to content

DB Helper

Thin database connectors. Narrative guide: DB Helper.

Interfaces

pypepper.helper.db.interfaces

Database helper configuration interfaces.

IConfig

Source code in pypepper/helper/db/interfaces.py 
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
class IConfig(metaclass=ABCMeta):
    uri: str | None
    username: str | None
    password: str | None
    host: str | None
    port: int = 0
    db: str | None

    def __init__(
        self,
        uri: str | None = None,
        username: str | None = None,
        password: str | None = None,
        host: str | None = None,
        port: int = 0,
        db: str | None = None,
    ):
        self.uri = uri
        self.username = username
        self.password = password
        self.host = host
        self.port = port
        self.db = db

uri instance-attribute 

uri: str | None = uri

username instance-attribute 

username: str | None = username

password instance-attribute 

password: str | None = password

host instance-attribute 

host: str | None = host

port class-attribute instance-attribute 

port: int = port

db instance-attribute 

db: str | None = db

MySQL

pypepper.helper.db.mysql

MySQL SQLAlchemy connector helper.

Config

Bases: IConfig

Source code in pypepper/helper/db/mysql.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
class Config(IConfig, metaclass=ABCMeta):
    charset: str = "utf8mb4"

    def __init__(
        self,
        uri: str | None = None,
        username: str | None = None,
        password: str | None = None,
        host: str | None = None,
        port: int = 3306,
        db: str | None = None,
        charset: str = "utf8mb4",
    ):
        super().__init__(
            uri=uri,
            username=username,
            password=password,
            host=host,
            port=port,
            db=db,
        )
        self.charset = charset

charset class-attribute instance-attribute 

charset: str = charset

connect 

connect(cfg: Config) -> Connection
Source code in pypepper/helper/db/mysql.py 
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
def connect(cfg: Config) -> Connection:
    if not cfg:
        raise ValueError("invalid database config")

    if cfg.uri:
        return create_engine(cfg.uri).connect()

    if not cfg.username:
        raise ValueError("invalid username")
    if not cfg.password:
        raise ValueError("invalid password")
    if not cfg.host:
        raise ValueError("invalid host")
    if not cfg.db:
        raise ValueError("invalid db")

    uri = build_mysql_uri(
        username=cfg.username,
        password=cfg.password,
        host=cfg.host,
        port=cfg.port,
        db=cfg.db,
        charset=cfg.charset,
    )
    return create_engine(uri).connect()

ping 

ping(engine: Connection) -> bool
Source code in pypepper/helper/db/mysql.py 
64
65
66
67
def ping(engine: Connection) -> bool:
    if not engine:
        raise ValueError("invalid engine")
    return not engine.closed

PostgreSQL

pypepper.helper.db.postgres

PostgreSQL SQLAlchemy connector helper.

Config

Bases: IConfig

Source code in pypepper/helper/db/postgres.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
class Config(IConfig, metaclass=ABCMeta):
    sslmode: str | None = None

    def __init__(
        self,
        uri: str | None = None,
        username: str | None = None,
        password: str | None = None,
        host: str | None = None,
        port: int = 5432,
        db: str | None = None,
        sslmode: str | None = None,
    ):
        super().__init__(
            uri=uri,
            username=username,
            password=password,
            host=host,
            port=port,
            db=db,
        )
        self.sslmode = sslmode

sslmode class-attribute instance-attribute 

sslmode: str | None = sslmode

connect 

connect(cfg: Config) -> Connection
Source code in pypepper/helper/db/postgres.py 
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
def connect(cfg: Config) -> Connection:
    if not cfg:
        raise ValueError("invalid database config")

    if cfg.uri:
        return create_engine(cfg.uri).connect()

    if not cfg.username:
        raise ValueError("invalid username")
    if not cfg.password:
        raise ValueError("invalid password")
    if not cfg.host:
        raise ValueError("invalid host")
    if not cfg.db:
        raise ValueError("invalid db")

    uri = build_postgres_uri(
        username=cfg.username,
        password=cfg.password,
        host=cfg.host,
        port=cfg.port,
        db=cfg.db,
        sslmode=cfg.sslmode,
    )
    return create_engine(uri).connect()

ping 

ping(engine: Connection) -> bool
Source code in pypepper/helper/db/postgres.py 
64
65
66
67
def ping(engine: Connection) -> bool:
    if not engine:
        raise ValueError("invalid engine")
    return not engine.closed

MongoDB

pypepper.helper.db.mongodb

MongoDB / mongoengine connector helper.

Config

Bases: IConfig

Source code in pypepper/helper/db/mongodb.py 
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
class Config(IConfig, metaclass=ABCMeta):
    auth_source: str = "admin"

    # Configuring a UUID Representation
    # Ref 1: https://github.com/MongoEngine/mongoengine/issues/2728
    # Ref 2: https://pymongo.readthedocs.io/en/stable/examples/uuid.html#configuring-uuid-representation
    uuid_representation: str = "standard"

    def __init__(
        self,
        uri: str | None = None,
        username: str | None = None,
        password: str | None = None,
        host: str | None = None,
        port: int = 27017,
        db: str | None = None,
        auth_source: str = "admin",
        uuid_representation: str = "standard",
    ):
        super().__init__(
            uri=uri,
            username=username,
            password=password,
            host=host,
            port=port,
            db=db,
        )
        self.auth_source = auth_source
        self.uuid_representation = uuid_representation

auth_source class-attribute instance-attribute 

auth_source: str = auth_source

uuid_representation class-attribute instance-attribute 

uuid_representation: str = uuid_representation

connect 

connect(cfg: Config) -> None
Source code in pypepper/helper/db/mongodb.py 
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
def connect(cfg: Config) -> None:
    if not cfg:
        raise ValueError("invalid database config")

    if cfg.uri:
        connect_db(
            host=cfg.uri,
            uuidRepresentation="standard",
        )
    else:
        connect_db(
            username=cfg.username,
            password=cfg.password,
            host=cfg.host,
            port=cfg.port,
            db=cfg.db,
            authentication_source=cfg.auth_source,
            uuidRepresentation="standard",
        )

close 

close()
Source code in pypepper/helper/db/mongodb.py 
65
66
def close():
    disconnect_all()