Ability to pass credentials by string

Author: vgvoleg

ydb_dbapi/connections.py

@@ -20,6 +20,7 @@
 from .errors import NotSupportedError
 from .utils import handle_ydb_errors
 from .utils import maybe_get_current_trace_id
+from .utils import prepare_credentials
 
 
 class IsolationLevel(str, Enum):
@@ -69,13 +70,13 @@ def __init__(
         port: str = "",
         database: str = "",
         ydb_table_path_prefix: str = "",
-        credentials: ydb.AbstractCredentials | None = None,
+        credentials: ydb.Credentials | dict | str | None = None,
         ydb_session_pool: SessionPool | AsyncSessionPool | None = None,
         **kwargs: dict,
     ) -> None:
-        self.endpoint = f"grpc://{host}:{port}"
+        self.endpoint = f"grpcs://{host}:{port}"
         self.database = database
-        self.credentials = credentials
+        self.credentials = prepare_credentials(credentials)
         self.table_path_prefix = ydb_table_path_prefix
 
         self.connection_kwargs: dict = kwargs
@@ -170,7 +171,7 @@ def __init__(
         port: str = "",
         database: str = "",
         ydb_table_path_prefix: str = "",
-        credentials: ydb.AbstractCredentials | None = None,
+        credentials: ydb.Credentials | None = None,
         ydb_session_pool: SessionPool | AsyncSessionPool | None = None,
         **kwargs: dict,
     ) -> None:
@@ -333,7 +334,7 @@ def __init__(
         port: str = "",
         database: str = "",
         ydb_table_path_prefix: str = "",
-        credentials: ydb.AbstractCredentials | None = None,
+        credentials: ydb.Credentials | None = None,
         ydb_session_pool: SessionPool | AsyncSessionPool | None = None,
         **kwargs: dict,
     ) -> None:

ydb_dbapi/utils.py

@@ -2,6 +2,7 @@
 
 import functools
 import importlib.util
+import json
 from enum import Enum
 from inspect import iscoroutinefunction
 from typing import Any
@@ -117,3 +118,30 @@ def maybe_get_current_trace_id() -> str | None:
 
     # Return None if OpenTelemetry is not available or trace ID is invalid
     return None
+
+
+def prepare_credentials(
+    credentials: ydb.Credentials | dict | str | None,
+) -> ydb.Credentials | None:
+    if not credentials:
+        return None
+
+    if isinstance(credentials, ydb.Credentials):
+        return credentials
+
+    if isinstance(credentials, str):
+        credentials = json.loads(credentials)
+
+    if isinstance(credentials, dict):
+        credentials = credentials or {}
+        token = credentials.get("token")
+        if token:
+            return ydb.AccessTokenCredentials(token)
+
+        service_account_json = credentials.get("service_account_json")
+        if service_account_json:
+            return ydb.iam.ServiceAccountCredentials.from_content(
+                json.dumps(service_account_json)
+            )
+
+    return ydb.AnonymousCredentials()