Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def _apply_mixins(self):
"""
apply mixins in defaults.OMEGA_MDF_APPLY_MIXINS
"""
from omegaml import settings
defaults = settings()
for mixin, applyto in defaults.OMEGA_MDF_APPLY_MIXINS:
if any(v in self.caller._applyto for v in applyto.split(',')):
extend_instance(self, mixin)
def _apply_mixins(self, *args, **kwargs):
"""
apply mixins in defaults.OMEGA_MDF_MIXINS
"""
from omegaml import settings
defaults = settings()
for mixin, applyto in defaults.OMEGA_MDF_MIXINS:
if any(v in self._applyto for v in applyto.split(',')):
extend_instance(self, mixin, *args, **kwargs)
def get_user_config_from_api(api_auth, api_url=None, requested_userid=None, view=False):
# safe way to talk to either the remote API or the in-process test server
from omegaml import settings
defaults = settings()
api_url = api_url or defaults.OMEGA_RESTAPI_URL
api_url += '/api/v1/config/'
api_url = api_url.replace('//api', '/api')
query = []
if requested_userid:
query.append('user={}'.format(requested_userid))
if view:
query.append('view={}'.format(int(view)))
api_url += '?' + '&'.join(query)
# -- setup appropriate client API
if api_url.startswith('http'):
import requests
server = requests
server_kwargs = dict(auth=api_auth)
deserialize = lambda resp: resp.json()
elif any('test' in v for v in sys.argv):
"""
setup an Omega instance from userid and apikey
:param userid: the userid
:param apikey: the apikey
:param api_url: the api URL
:param requested_userid: the userid to request config for. in this case userid
and apikey must for a staff user for the request to succeed
:param qualifier: the database qualifier requested. defaults to 'default'
:returns: the Omega instance configured for the given user
"""
from omegaml import Omega
from omegaml import settings, _base_config
from omegaml.util import DefaultsContext
defaults = DefaultsContext(settings())
qualifier = qualifier or 'default'
api_url = api_url or defaults.OMEGA_RESTAPI_URL
if api_url.startswith('http') or any('test' in v for v in sys.argv):
api_auth = OmegaRestApiAuth(userid, apikey)
configs = get_user_config_from_api(api_auth, api_url=api_url,
requested_userid=requested_userid,
view=view)
configs = configs['objects'][0]['data']
elif api_url == 'local':
configs = {k: getattr(defaults, k) for k in dir(defaults) if k.startswith('OMEGA')}
else:
raise ValueError('invalid api_url {}'.format(api_url))
if qualifier == 'default':
config = configs.get(qualifier, configs)
else:
config = configs[qualifier]
def save_userconfig_from_apikey(configfile, userid, apikey, api_url=None, requested_userid=None,
view=False):
from omegaml import settings
defaults = settings()
api_url = api_url or defaults.OMEGA_RESTAPI_URL
with open(configfile, 'w') as fconfig:
auth = OmegaRestApiAuth(userid, apikey)
configs = get_user_config_from_api(auth,
api_url=api_url,
requested_userid=requested_userid,
view=view)
config = configs['objects'][0]['data']
config['OMEGA_RESTAPI_URL'] = api_url
yaml.safe_dump(config, fconfig, default_flow_style=False)
print("Config is in {configfile}".format(**locals()))
def setup(app):
from omegaml import settings
app.add_stylesheet('custom.css')
del os.environ['DJANGO_SETTINGS_MODULE']
defaults = settings()
def apply_mixins(self):
"""
apply mixins in defaults.OMEGA_RUNTIME_MIXINS
"""
from omegaml import settings
defaults = settings()
for mixin in defaults.OMEGA_RUNTIME_MIXINS:
extend_instance(self, mixin)