Skip to content

Oracle

To connect to an Oracle database, is possible to import the OracleConnector class from the hamana.connector.db.oracle module. As a shortcut, the OracleConnector class is also available in the hamana.connector.db module as Oracle.

import hamana as hm

# connect database
oracle_db = hm.connector.db.Oracle.new(
    host = "localhost",
    port = 1521,
    user = "user",
    password = "password"
)

# perform operations
# ...

hamana.connector.db.oracle

OracleConnectorConfig dataclass

OracleConnectorConfig(
    host: str | None = None,
    port: int = 1521,
    user: str | None = None,
    password: str | None = None,
    service: str | None = None,
    data_source_name: str | None = None,
)

Bases: DatabaseConnectorConfig

Class to represent the configuration of an Oracle database.

port class-attribute instance-attribute

port: int = 1521

Port of the database. Default is 1521.

service class-attribute instance-attribute

service: str | None = None

Service name of the database.

data_source_name class-attribute instance-attribute

data_source_name: str | None = None

DSN connection string to connect on the database.

connect_params property

connect_params: ConnectParams

Get the connection parameters to connect on the database.

get_data_source_name

get_data_source_name() -> str

Get the DSN connection string to connect on the database.

Source code in src/hamana/connector/db/oracle.py
42
43
44
def get_data_source_name(self) -> str:
    """Get the DSN connection string to connect on the database."""
    return self.data_source_name if self.data_source_name else self.connect_params.get_connect_string()

OracleConnector

OracleConnector(
    config: OracleConnectorConfig, **kwargs: dict[str, Any]
)

Bases: BaseConnector

Class representing a connector to an Oracle database.

Source code in src/hamana/connector/db/oracle.py
52
53
54
55
def __init__(self, config: OracleConnectorConfig, **kwargs: dict[str, Any]) -> None:
    self.config = config
    self.kwargs = kwargs
    self.connection: Connection

create_config classmethod

create_config(
    user: str,
    password: str,
    host: str | None = None,
    service: str | None = None,
    port: int = 1521,
    data_source_name: str | None = None,
) -> OracleConnectorConfig

Use this function to create a new Oracle connector configuration.

Parameters:

Name Type Description Default
user str

User to connect to the database.

required
password str

Password to uso to connect to the database.

required
host str | None

Host of the database.

None
service str | None

Service name of the database.

None
port int

Port of the database.

1521
data_source_name str | None

DSN connection string to connect on the database. Observe that if DSN is provided, then host, service and port are ignored.

None

Returns:

Type Description
OracleConnectorConfig

Oracle connector configuration.

Source code in src/hamana/connector/db/oracle.py
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
@classmethod
def create_config(
    cls,
    user: str,
    password: str,
    host: str | None = None,
    service: str | None = None,
    port: int = 1521,
    data_source_name: str | None = None
) -> OracleConnectorConfig:
    """
        Use this function to create a new Oracle connector configuration.

        Parameters:
            user: User to connect to the database.
            password: Password to uso to connect to the database.
            host: Host of the database.
            service: Service name of the database.
            port: Port of the database.
            data_source_name: DSN connection string to connect on the database. 
                Observe that if DSN is provided, then host, service and port are ignored.

        Returns:
            Oracle connector configuration.
    """
    logger.debug("start")

    if data_source_name:
        logger.debug("data source name provided")
        config = OracleConnectorConfig(user = user, password = password, data_source_name = data_source_name)
    else:
        logger.debug(f"host: {host}, service: {service}, port: {port}")
        config = OracleConnectorConfig(user = user, password = password, host = host, service = service, port = port)

    logger.debug("end")
    return config

new classmethod

new(
    user: str,
    password: str,
    host: str | None = None,
    service: str | None = None,
    port: int = 1521,
    data_source_name: str | None = None,
) -> "OracleConnector"

Use this function to create a new Oracle connector.

Parameters:

Name Type Description Default
user str

User to connect to the database.

required
password str

Password to uso to connect to the database.

required
host str | None

Host of the database.

None
service str | None

Service name of the database.

None
port int

Port of the database.

1521
data_source_name str | None

DSN connection string to connect on the database. Observe that if DSN is provided, then host, service and port are ignored.

None

Returns:

Type Description
'OracleConnector'

Oracle connector.

Source code in src/hamana/connector/db/oracle.py
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
@classmethod
def new(
    cls,
    user: str,
    password: str,
    host: str | None = None,
    service: str | None = None,
    port: int = 1521,
    data_source_name: str | None = None
) -> "OracleConnector":
    """
        Use this function to create a new Oracle connector.

        Parameters:
            user: User to connect to the database.
            password: Password to uso to connect to the database.
            host: Host of the database.
            service: Service name of the database.
            port: Port of the database.
            data_source_name: DSN connection string to connect on the database. 
                Observe that if DSN is provided, then host, service and port are ignored.

        Returns:
            Oracle connector.
    """
    logger.debug("start")
    config = cls.create_config(
        user = user,
        password = password,
        host = host,
        service = service,
        port = port,
        data_source_name = data_source_name
    )
    logger.debug("end")
    return cls(config)