Skip to content

Commit 3dd11f7

Browse files
kevinjqliukevinjqliu
committed
rename get
1 parent 7c48350 commit 3dd11f7

File tree

1 file changed

+51
-47
lines changed

1 file changed

+51
-47
lines changed

‎pyiceberg/io/pyarrow.py‎

Lines changed: 51 additions & 47 deletions
Original file line numberDiff line numberDiff line change
@@ -468,38 +468,38 @@ def _initialize_oss_fs(self) -> FileSystem:
468468
properties = filter_properties(self.properties, key_predicate=lambda k: k.startswith(("s3.", "client.")))
469469
used_keys: set[str] = set()
470470

471-
def get(*keys: str) -> str | None:
471+
def get_property_with_tracking(*keys: str) -> str | None:
472472
return get_first_property_value_with_tracking(properties, used_keys, *keys)
473473

474474
client_kwargs: Properties = {}
475475

476-
if endpoint := get(S3_ENDPOINT, "s3.endpoint_override"):
476+
if endpoint := get_property_with_tracking(S3_ENDPOINT, "s3.endpoint_override"):
477477
client_kwargs["endpoint_override"] = endpoint
478-
if access_key := get(S3_ACCESS_KEY_ID, AWS_ACCESS_KEY_ID, "s3.access_key"):
478+
if access_key := get_property_with_tracking(S3_ACCESS_KEY_ID, AWS_ACCESS_KEY_ID, "s3.access_key"):
479479
client_kwargs["access_key"] = access_key
480-
if secret_key := get(S3_SECRET_ACCESS_KEY, AWS_SECRET_ACCESS_KEY, "s3.secret_key"):
480+
if secret_key := get_property_with_tracking(S3_SECRET_ACCESS_KEY, AWS_SECRET_ACCESS_KEY, "s3.secret_key"):
481481
client_kwargs["secret_key"] = secret_key
482-
if session_token := get(S3_SESSION_TOKEN, AWS_SESSION_TOKEN, "s3.session_token"):
482+
if session_token := get_property_with_tracking(S3_SESSION_TOKEN, AWS_SESSION_TOKEN, "s3.session_token"):
483483
client_kwargs["session_token"] = session_token
484-
if region := get(S3_REGION, AWS_REGION):
484+
if region := get_property_with_tracking(S3_REGION, AWS_REGION):
485485
client_kwargs["region"] = region
486-
_ = get(
486+
_ = get_property_with_tracking(
487487
S3_RESOLVE_REGION
488488
) # this feature is only available for S3. Use `get` here so it does not get passed down to the S3FileSystem constructor
489-
if force_virtual_addressing := get(S3_FORCE_VIRTUAL_ADDRESSING, "s3.force_virtual_addressing"):
489+
if force_virtual_addressing := get_property_with_tracking(S3_FORCE_VIRTUAL_ADDRESSING, "s3.force_virtual_addressing"):
490490
client_kwargs["force_virtual_addressing"] = convert_str_to_bool(force_virtual_addressing)
491491
else:
492492
# For Alibaba OSS protocol, default to True
493493
client_kwargs["force_virtual_addressing"] = True
494-
if proxy_uri := get(S3_PROXY_URI, "s3.proxy_options"):
494+
if proxy_uri := get_property_with_tracking(S3_PROXY_URI, "s3.proxy_options"):
495495
client_kwargs["proxy_options"] = proxy_uri
496-
if connect_timeout := get(S3_CONNECT_TIMEOUT, "s3.connect_timeout"):
496+
if connect_timeout := get_property_with_tracking(S3_CONNECT_TIMEOUT, "s3.connect_timeout"):
497497
client_kwargs["connect_timeout"] = float(connect_timeout)
498-
if request_timeout := get(S3_REQUEST_TIMEOUT, "s3.request_timeout"):
498+
if request_timeout := get_property_with_tracking(S3_REQUEST_TIMEOUT, "s3.request_timeout"):
499499
client_kwargs["request_timeout"] = float(request_timeout)
500-
if role_arn := get(S3_ROLE_ARN, AWS_ROLE_ARN, "s3.role_arn"):
500+
if role_arn := get_property_with_tracking(S3_ROLE_ARN, AWS_ROLE_ARN, "s3.role_arn"):
501501
client_kwargs["role_arn"] = role_arn
502-
if session_name := get(S3_ROLE_SESSION_NAME, AWS_ROLE_SESSION_NAME, "s3.session_name"):
502+
if session_name := get_property_with_tracking(S3_ROLE_SESSION_NAME, AWS_ROLE_SESSION_NAME, "s3.session_name"):
503503
client_kwargs["session_name"] = session_name
504504

505505
# get the rest of the properties with the `s3.` prefix that are not already evaluated
@@ -513,39 +513,41 @@ def _initialize_s3_fs(self, netloc: Optional[str]) -> FileSystem:
513513
properties = filter_properties(self.properties, key_predicate=lambda k: k.startswith(("s3.", "client.")))
514514
used_keys: set[str] = set()
515515

516-
def get(*keys: str) -> str | None:
516+
def get_property_with_tracking(*keys: str) -> str | None:
517517
return get_first_property_value_with_tracking(properties, used_keys, *keys)
518518

519519
client_kwargs: Properties = {}
520520

521521
# Handle S3 region configuration with optional auto-resolution
522522
client_kwargs["region"] = self._resolve_s3_region(
523-
provided_region=get(S3_REGION, AWS_REGION), resolve_region_override=get(S3_RESOLVE_REGION), bucket=netloc
523+
provided_region=get_property_with_tracking(S3_REGION, AWS_REGION),
524+
resolve_region_override=get_property_with_tracking(S3_RESOLVE_REGION),
525+
bucket=netloc,
524526
)
525527

526-
if endpoint := get(S3_ENDPOINT, "s3.endpoint_override"):
528+
if endpoint := get_property_with_tracking(S3_ENDPOINT, "s3.endpoint_override"):
527529
client_kwargs["endpoint_override"] = endpoint
528-
if access_key := get(S3_ACCESS_KEY_ID, AWS_ACCESS_KEY_ID, "s3.access_key"):
530+
if access_key := get_property_with_tracking(S3_ACCESS_KEY_ID, AWS_ACCESS_KEY_ID, "s3.access_key"):
529531
client_kwargs["access_key"] = access_key
530-
if secret_key := get(S3_SECRET_ACCESS_KEY, AWS_SECRET_ACCESS_KEY, "s3.secret_key"):
532+
if secret_key := get_property_with_tracking(S3_SECRET_ACCESS_KEY, AWS_SECRET_ACCESS_KEY, "s3.secret_key"):
531533
client_kwargs["secret_key"] = secret_key
532-
if session_token := get(S3_SESSION_TOKEN, AWS_SESSION_TOKEN, "s3.session_token"):
534+
if session_token := get_property_with_tracking(S3_SESSION_TOKEN, AWS_SESSION_TOKEN, "s3.session_token"):
533535
client_kwargs["session_token"] = session_token
534-
if proxy_uri := get(S3_PROXY_URI, "s3.proxy_options"):
536+
if proxy_uri := get_property_with_tracking(S3_PROXY_URI, "s3.proxy_options"):
535537
client_kwargs["proxy_options"] = proxy_uri
536-
if connect_timeout := get(S3_CONNECT_TIMEOUT, "s3.connect_timeout"):
538+
if connect_timeout := get_property_with_tracking(S3_CONNECT_TIMEOUT, "s3.connect_timeout"):
537539
client_kwargs["connect_timeout"] = float(connect_timeout)
538-
if request_timeout := get(S3_REQUEST_TIMEOUT, "s3.request_timeout"):
540+
if request_timeout := get_property_with_tracking(S3_REQUEST_TIMEOUT, "s3.request_timeout"):
539541
client_kwargs["request_timeout"] = float(request_timeout)
540-
if role_arn := get(S3_ROLE_ARN, AWS_ROLE_ARN, "s3.role_arn"):
542+
if role_arn := get_property_with_tracking(S3_ROLE_ARN, AWS_ROLE_ARN, "s3.role_arn"):
541543
client_kwargs["role_arn"] = role_arn
542-
if session_name := get(S3_ROLE_SESSION_NAME, AWS_ROLE_SESSION_NAME, "s3.session_name"):
544+
if session_name := get_property_with_tracking(S3_ROLE_SESSION_NAME, AWS_ROLE_SESSION_NAME, "s3.session_name"):
543545
client_kwargs["session_name"] = session_name
544546

545-
if force_virtual_addressing := get(S3_FORCE_VIRTUAL_ADDRESSING, "s3.force_virtual_addressing"):
547+
if force_virtual_addressing := get_property_with_tracking(S3_FORCE_VIRTUAL_ADDRESSING, "s3.force_virtual_addressing"):
546548
client_kwargs["force_virtual_addressing"] = convert_str_to_bool(force_virtual_addressing)
547549
# Handle retry strategy special case
548-
if retry_strategy_impl := get(S3_RETRY_STRATEGY_IMPL, "s3.retry_strategy"):
550+
if retry_strategy_impl := get_property_with_tracking(S3_RETRY_STRATEGY_IMPL, "s3.retry_strategy"):
549551
if retry_instance := _import_retry_strategy(retry_strategy_impl):
550552
client_kwargs["retry_strategy"] = retry_instance
551553

@@ -569,30 +571,30 @@ def _initialize_azure_fs(self) -> FileSystem:
569571
properties = filter_properties(self.properties, key_predicate=lambda k: k.startswith("adls."))
570572
used_keys: set[str] = set()
571573

572-
def get(*keys: str) -> str | None:
574+
def get_property_with_tracking(*keys: str) -> str | None:
573575
return get_first_property_value_with_tracking(properties, used_keys, *keys)
574576

575577
client_kwargs: Properties = {}
576578

577-
if account_name := get(ADLS_ACCOUNT_NAME, "adls.account_name"):
579+
if account_name := get_property_with_tracking(ADLS_ACCOUNT_NAME, "adls.account_name"):
578580
client_kwargs["account_name"] = account_name
579581

580-
if account_key := get(ADLS_ACCOUNT_KEY, "adls.account_key"):
582+
if account_key := get_property_with_tracking(ADLS_ACCOUNT_KEY, "adls.account_key"):
581583
client_kwargs["account_key"] = account_key
582584

583-
if blob_storage_authority := get(ADLS_BLOB_STORAGE_AUTHORITY, "adls.blob_storage_authority"):
585+
if blob_storage_authority := get_property_with_tracking(ADLS_BLOB_STORAGE_AUTHORITY, "adls.blob_storage_authority"):
584586
client_kwargs["blob_storage_authority"] = blob_storage_authority
585587

586-
if dfs_storage_authority := get(ADLS_DFS_STORAGE_AUTHORITY, "adls.dfs_storage_authority"):
588+
if dfs_storage_authority := get_property_with_tracking(ADLS_DFS_STORAGE_AUTHORITY, "adls.dfs_storage_authority"):
587589
client_kwargs["dfs_storage_authority"] = dfs_storage_authority
588590

589-
if blob_storage_scheme := get(ADLS_BLOB_STORAGE_SCHEME, "adls.blob_storage_scheme"):
591+
if blob_storage_scheme := get_property_with_tracking(ADLS_BLOB_STORAGE_SCHEME, "adls.blob_storage_scheme"):
590592
client_kwargs["blob_storage_scheme"] = blob_storage_scheme
591593

592-
if dfs_storage_scheme := get(ADLS_DFS_STORAGE_SCHEME, "adls.dfs_storage_scheme"):
594+
if dfs_storage_scheme := get_property_with_tracking(ADLS_DFS_STORAGE_SCHEME, "adls.dfs_storage_scheme"):
593595
client_kwargs["dfs_storage_scheme"] = dfs_storage_scheme
594596

595-
if sas_token := get(ADLS_SAS_TOKEN, "adls.sas_token"):
597+
if sas_token := get_property_with_tracking(ADLS_SAS_TOKEN, "adls.sas_token"):
596598
client_kwargs["sas_token"] = sas_token
597599

598600
# get the rest of the properties with the `adls.` prefix that are not already evaluated
@@ -609,19 +611,19 @@ def _initialize_hdfs_fs(self, scheme: str, netloc: Optional[str]) -> FileSystem:
609611
properties = filter_properties(self.properties, key_predicate=lambda k: k.startswith("hdfs."))
610612
used_keys: set[str] = set()
611613

612-
def get(*keys: str) -> str | None:
614+
def get_property_with_tracking(*keys: str) -> str | None:
613615
return get_first_property_value_with_tracking(properties, used_keys, *keys)
614616

615617
client_kwargs: Properties = {}
616618

617-
if host := get(HDFS_HOST):
619+
if host := get_property_with_tracking(HDFS_HOST):
618620
client_kwargs["host"] = host
619-
if port := get(HDFS_PORT):
621+
if port := get_property_with_tracking(HDFS_PORT):
620622
# port should be an integer type
621623
client_kwargs["port"] = int(port)
622-
if user := get(HDFS_USER):
624+
if user := get_property_with_tracking(HDFS_USER):
623625
client_kwargs["user"] = user
624-
if kerb_ticket := get(HDFS_KERB_TICKET, "hdfs.kerb_ticket"):
626+
if kerb_ticket := get_property_with_tracking(HDFS_KERB_TICKET, "hdfs.kerb_ticket"):
625627
client_kwargs["kerb_ticket"] = kerb_ticket
626628

627629
# get the rest of the properties with the `hdfs.` prefix that are not already evaluated
@@ -635,29 +637,31 @@ def _initialize_gcs_fs(self) -> FileSystem:
635637
properties = filter_properties(self.properties, key_predicate=lambda k: k.startswith("gcs."))
636638
used_keys: set[str] = set()
637639

638-
def get(*keys: str) -> str | None:
640+
def get_property_with_tracking(*keys: str) -> str | None:
639641
return get_first_property_value_with_tracking(properties, used_keys, *keys)
640642

641643
client_kwargs: Properties = {}
642644

643-
if access_token := get(GCS_TOKEN, "gcs.access_token"):
645+
if access_token := get_property_with_tracking(GCS_TOKEN, "gcs.access_token"):
644646
client_kwargs["access_token"] = access_token
645-
if expiration := get(GCS_TOKEN_EXPIRES_AT_MS, "gcs.credential_token_expiration"):
647+
if expiration := get_property_with_tracking(GCS_TOKEN_EXPIRES_AT_MS, "gcs.credential_token_expiration"):
646648
client_kwargs["credential_token_expiration"] = millis_to_datetime(int(expiration))
647-
if bucket_location := get(GCS_DEFAULT_LOCATION, "gcs.default_bucket_location"):
649+
if bucket_location := get_property_with_tracking(GCS_DEFAULT_LOCATION, "gcs.default_bucket_location"):
648650
client_kwargs["default_bucket_location"] = bucket_location
649-
if endpoint := get(GCS_SERVICE_HOST):
651+
if endpoint := get_property_with_tracking(GCS_SERVICE_HOST):
650652
url_parts = urlparse(endpoint)
651653
client_kwargs["scheme"] = url_parts.scheme
652654
client_kwargs["endpoint_override"] = url_parts.netloc
653-
if (scheme := get("gcs.scheme")) and "scheme" not in client_kwargs: # GCS_SERVICE_HOST takes precedence
655+
if (
656+
scheme := get_property_with_tracking("gcs.scheme")
657+
) and "scheme" not in client_kwargs: # GCS_SERVICE_HOST takes precedence
654658
client_kwargs["scheme"] = scheme
655659
if (
656-
endpoint_override := get("gcs.endpoint_override")
660+
endpoint_override := get_property_with_tracking("gcs.endpoint_override")
657661
) and "endpoint_override" not in client_kwargs: # GCS_SERVICE_HOST takes precedence
658662
client_kwargs["endpoint_override"] = endpoint_override
659663

660-
if project_id := get(GCS_PROJECT_ID, "gcs.project_id"):
664+
if project_id := get_property_with_tracking(GCS_PROJECT_ID, "gcs.project_id"):
661665
client_kwargs["project_id"] = project_id
662666

663667
# get the rest of the properties with the `gcs.` prefix that are not already evaluated

0 commit comments

Comments
 (0)