-
Notifications
You must be signed in to change notification settings - Fork 590
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #2339 from vaexio/fix_ci2
ci: fix CI, use macOS-11 and remove pydantic
- Loading branch information
Showing
10 changed files
with
203 additions
and
53 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -1,16 +1,7 @@ | ||
import os | ||
from pydantic import BaseSettings | ||
import vaex.utils | ||
import pydantic | ||
|
||
|
||
# in seperate module to avoid circular imports | ||
|
||
class ConfigDefault: | ||
env_file = ".env" | ||
@classmethod | ||
def customise_sources(cls, init_settings, env_settings, file_secret_settings): | ||
return ( | ||
env_settings, | ||
file_secret_settings, | ||
init_settings, # constructor argument last, since they come from global yaml file | ||
) | ||
pass |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,130 @@ | ||
import os | ||
from pathlib import Path | ||
from typing import Any, Optional | ||
|
||
# similar API to pydantic/pydantic-settings but we prefer not to have a dependency on pydantic | ||
# since we cannot be compatible with pydantic1 and 2 | ||
# NOTE: not a public api | ||
|
||
|
||
def _get_type(annotation): | ||
check_optional_types = [str, int, float, bool, dict, list] | ||
for check_type in check_optional_types: | ||
if annotation == Optional[check_type]: | ||
return check_type | ||
if hasattr(annotation, "__origin__"): | ||
if annotation.__origin__ == dict: | ||
return dict | ||
return annotation | ||
|
||
|
||
class _Field: | ||
def __init__(self, default=None, env=None, title=None, default_factory=None, gt=None, alias=None) -> None: | ||
self.default = default | ||
self.env = env | ||
self.fullenv = None | ||
self.title = title | ||
self.annotation = None | ||
self.default_factory = default_factory | ||
self.gt = gt | ||
self.alias = alias | ||
self.field_info = self | ||
self.extra = {"env_names": [env] if env else []} | ||
|
||
def __set_name__(self, owner, name): | ||
prefix = "SOLARA_" | ||
config = getattr(owner, "Config") | ||
if config: | ||
prefix = getattr(config, "env_prefix", prefix).upper() | ||
if hasattr(config, "fields"): | ||
fields = config.fields | ||
if name in fields: | ||
self.alias = fields[name] | ||
self.name = name | ||
self.alias = self.alias or self.name | ||
self.title = self.title or self.name | ||
if self.env is None: | ||
self.env = f"{prefix}{self.name.upper()}" | ||
else: | ||
self.env = self.env | ||
self.annotation = owner.__annotations__.get(self.name) | ||
assert self.annotation is not None, f"Field {self.name} must have a type annotation" | ||
self.type_ = _get_type(self.annotation) | ||
|
||
def __get__(self, instance, owner): | ||
if instance is None: | ||
return self | ||
return instance._values[self.name] | ||
|
||
|
||
def convert(annotation, value: str) -> Any: | ||
check_optional_types = [str, int, float, bool, Path] | ||
for check_type in check_optional_types: | ||
if annotation == Optional[check_type]: | ||
annotation = check_type | ||
return convert(annotation, value) | ||
if annotation == str: | ||
return value | ||
elif annotation == int: | ||
return int(value) | ||
elif annotation == float: | ||
return float(value) | ||
elif annotation == bool: | ||
if value in ("True", "true", "1"): | ||
return True | ||
elif value in ("False", "false", "0"): | ||
return False | ||
else: | ||
raise ValueError(f"Invalid boolean value {value}") | ||
else: | ||
# raise TypeError(f"Unsupported type {annotation}") | ||
return annotation(value) | ||
|
||
|
||
def Field(*args, **kwargs) -> Any: | ||
return _Field(*args, **kwargs) | ||
|
||
|
||
class BaseSettings: | ||
__fields__: dict | ||
|
||
def __init__(self, **kwargs) -> None: | ||
cls = type(self) | ||
self._values = {**kwargs} | ||
keys = set([k.upper() for k in os.environ.keys()]) | ||
for key, field in cls.__dict__.items(): | ||
if key in kwargs: | ||
continue | ||
if isinstance(field, _Field): | ||
value = field.default | ||
if field.default_factory: | ||
value = field.default_factory() | ||
|
||
if field.env: | ||
env_key = field.env.upper() | ||
if env_key in keys: | ||
# do a case-insensitive lookup | ||
for env_var_cased in os.environ.keys(): | ||
if env_key.upper() == env_var_cased.upper(): | ||
value = convert(field.annotation, os.environ[env_var_cased]) | ||
self._values[key] = value | ||
|
||
def __init_subclass__(cls) -> None: | ||
cls.__fields__ = {} | ||
for key, field in cls.__dict__.items(): | ||
if key.startswith("_"): | ||
continue | ||
if key == "Config": | ||
continue | ||
if not isinstance(field, _Field): | ||
field = Field(field) | ||
setattr(cls, key, field) | ||
field.__set_name__(cls, key) | ||
cls.__fields__[key] = field | ||
|
||
def dict(self, by_alias=True): | ||
values = self._values.copy() | ||
for key, value in values.items(): | ||
if isinstance(value, BaseSettings): | ||
values[key] = value.dict(by_alias=by_alias) | ||
return values |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.