How to use the addict.Dict function in addict

To help you get started, we’ve selected a few addict examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github Yelp / paasta / tests / frameworks / test_adhoc_scheduler.py View on Github external
def make_fake_offer(
    cpu=50000, mem=50000, port_begin=31000, port_end=32000, pool="default"
):
    offer = Dict(
        agent_id=Dict(value="super_big_slave"),
        resources=[
            Dict(name="cpus", scalar=Dict(value=cpu)),
            Dict(name="mem", scalar=Dict(value=mem)),
            Dict(
                name="ports", ranges=Dict(range=[Dict(begin=port_begin, end=port_end)])
            ),
        ],
        attributes=[],
    )

    if pool is not None:
        offer.attributes = [Dict(name="pool", text=Dict(value=pool))]

    return offer
github Yelp / paasta / tests / frameworks / test_native_scheduler.py View on Github external
def make_fake_offer(
    cpu=50000, mem=50000, port_begin=31000, port_end=32000, pool="default"
):
    offer = Dict(
        agent_id=Dict(value="super_big_slave"),
        resources=[
            Dict(name="cpus", scalar=Dict(value=cpu)),
            Dict(name="mem", scalar=Dict(value=mem)),
            Dict(
                name="ports", ranges=Dict(range=[Dict(begin=port_begin, end=port_end)])
            ),
        ],
        attributes=[],
    )

    if pool is not None:
        offer.attributes = [Dict(name="pool", text=Dict(value=pool))]

    return offer
github goodrain / rainbond / hack / contrib / docker / chaos / plugins / clients / hubimageutils.py View on Github external
def parse_image(self, image):
        if '/' in image:
            host, full_name = image.split('/', 1)
        else:
            host, full_name = (None, image)
        if ':' in full_name:
            name, tag = full_name.split(':', 1)
        else:
            name, tag = (full_name, 'latest')
        return Dict(host=host, name=name, tag=tag)
github alin23 / spfy / spfy / result.py View on Github external
if "playlists" in self.result:
            data["context_uri"] = self.client._get_playlist_uri(item)
        elif "artists" in self.result:
            data["context_uri"] = self.client._get_artist_uri(item)
        elif "albums" in self.result:
            data["context_uri"] = self.client._get_album_uri(item)
        elif "items" in self.result:
            data["context_uri"] = self.client._get_uri(item.type, item)
        elif self.result.type and self.result.type in {"album", "artist", "playlist"}:
            data["context_uri"] = self.client._get_uri(self.result.type, self.result)
        elif item.type == "track":
            data["uris"] = list(map(self.client._get_track_uri, self.result))
        return data


class SpotifyResult(addict.Dict):
    ITER_KEYS = (
        "items",
        "artists",
        "tracks",
        "albums",
        "audio_features",
        "playlists",
        "devices",
    )

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self._next_result_available = threading.Event()
        self._next_result = None
        self._playable = Playable(self)
github Yelp / task_processing / task_processing / plugins / mesos / translator.py View on Github external
role=role,
            scalar=addict.Dict(value=task_config.mem)
        ),
        addict.Dict(
            name='disk',
            type='SCALAR',
            role=role,
            scalar=addict.Dict(value=task_config.disk)
        ),
        addict.Dict(
            name='gpus',
            type='SCALAR',
            role=role,
            scalar=addict.Dict(value=task_config.gpus)
        ),
        addict.Dict(
            name='ports',
            type='RANGES',
            role=role,
            ranges=addict.Dict(range=thaw(task_config.ports)),
        ),
github Yelp / task_processing / task_processing / plugins / mesos / translator.py View on Github external
def make_mesos_container_info(task_config: MesosTaskConfig) -> addict.Dict:
    container_info = addict.Dict(
        type=task_config.containerizer,
        volumes=thaw(task_config.volumes),
    )
    port_mappings = [addict.Dict(
        host_port=task_config.ports[0]['begin'], container_port=8888)]
    if container_info.type == 'DOCKER':
        container_info.docker = addict.Dict(
            image=task_config.image,
            network='BRIDGE',
            port_mappings=port_mappings,
            parameters=thaw(task_config.docker_parameters),
            force_pull_image=(not task_config.use_cached_image),
        )
    elif container_info.type == 'MESOS':
        container_info.network_infos = addict.Dict(port_mappings=port_mappings)
        # For this to work, image_providers needs to be set to 'docker' on mesos agents (as opposed
github douban / dpark / tools / scheduler.py View on Github external
def create_task(self, offer, t, k):
        task = Dict()
        task.task_id.value = '%s-%s' % (t.id, t.tried)
        task.agent_id.value = offer.agent_id.value
        task.name = 'task %s' % t.id
        task.executor = self.executor
        env = dict(os.environ)
        task.data = encode_data(pickle.dumps([
            os.getcwd(), None, env, self.options.shell,
            self.stdout_port, self.stderr_port, self.publisher_port
        ]))

        task.resources = resources = []
        cpu = Dict()
        resources.append(cpu)
        cpu.name = 'cpus'
        cpu.type = 'SCALAR'
        cpu.scalar.value = self.cpus * k
github opentargets / rest_api / app / common / elasticsearchclient.py View on Github external
def _get_exact_mapping_query(self, searchphrase):
        mm1 = addict.Dict()
        mm1.multi_match.query = searchphrase
        mm1.multi_match.fields = ["name^3",
                                  "description^2",
                                  "id",
                                  "approved_symbol",
                                  "symbol_synonyms",
                                  "name_synonyms",
                                  "uniprot_accessions",
                                  "hgnc_id",
                                  "ensembl_gene_id",
                                  "efo_path_codes",
                                  "efo_url",
                                  "efo_synonyms^0.5",
                                  "ortholog.*.symbol^0.5",
                                  "ortholog.*.id",
                                ]
github douban / pymesos / examples / scheduler.py View on Github external
def main(master):
    executor = Dict()
    executor.executor_id.value = 'MinimalExecutor'
    executor.name = executor.executor_id.value
    executor.command.value = '%s %s' % (
        sys.executable,
        abspath(join(dirname(__file__), 'executor.py'))
    )
    executor.resources = [
        dict(name='mem', type='SCALAR', scalar={'value': EXECUTOR_MEM}),
        dict(name='cpus', type='SCALAR', scalar={'value': EXECUTOR_CPUS}),
    ]

    framework = Dict()
    framework.user = getpass.getuser()
    framework.name = "MinimalFramework"
    framework.hostname = socket.gethostname()

    driver = MesosSchedulerDriver(
        MinimalScheduler(executor),
        framework,
        master,
        use_addict=True,
    )

    def signal_handler(signal, frame):
        driver.stop()

    def run_driver_thread():
        driver.run()
github NCI-GDC / gdcdatamodel / gdcdatamodel / mappings.py View on Github external
def _get_header(source):
    header = Dict()
    header.dynamic = 'strict'
    header._all.enabled = False
    header._source.compress = True
    header._source.excludes = ["__comment__"]
    header._id = {'path': '{}_id'.format(source)}
    return header

addict

Addict is a dictionary whose items can be set using both attribute and item syntax.

MIT
Latest version published 4 years ago

Package Health Score

76 / 100
Full package analysis