gidgid

from functools import singledispatch


@singledispatch  # 1
def to_json(inp):  # 1
    """implementations in dispatched functions"""


@to_json.register  # 2
def as_json_str(input_str: str) -> str:  # 2

gidgid

from typing import Union

from pydantic import BaseSettings, parse_obj_as
from typing_extensions import Literal


class LocalContext(BaseSettings):
    env: Literal["local"]  # 1
    mongo_url: str

gidgid

import os
from typing import Union

from pymongo import MongoClient, ReadPreference
import pytest
from functools import singledispatch
from pydantic import BaseSettings, parse_obj_as
from typing_extensions import Literal

gidgid

from dataclasses import dataclass
from typing import Iterable, Tuple, Dict, Set
from more_itertools import map_reduce


@dataclass(frozen=True, eq=True)
class Order:  # 1
    order_id: str  # 1
    price: int  # 1

gidgid

from typing import Iterable, Tuple, Dict, Set
from more_itertools import map_reduce


def group_edges(edges: Iterable[Tuple[str, str]]) -> Dict[str, Set[str]]:
    return map_reduce(
        edges,
        keyfunc=lambda edge: edge[0],  # 1
        valuefunc=lambda edge: edge[1],  # 2
        reducefunc=set,  # 3

gidgid

import pytest
from more_itertools import one
import networkx as nx


class GraphGenerationError(Exception):
    pass


def find_root(graph) -> str:

gidgid

from functools import reduce
from operator import add
import pytest
from more_itertools import always_iterable


def safer_sum(inputs) -> int:
    iterable_inputs = always_iterable(inputs)  # 1
    return reduce(add, iterable_inputs, 0)  # 1

gidgid

from more_itertools import always_iterable

always_iterable([1,3, 5])  # => <list_iterator at 0x107deca30>
list(always_iterable([1,3, 5]))  # => [1, 3, 5]
list(always_iterable({1,3, 5}))  # => [1, 3, 5] (works on all iterables)
list(always_iterable(42))  # => [42]
list(always_iterable([]))  # => []
list(always_iterable(None))  # => []

gidgid

from typing import Set
from dataclasses import dataclass
from itertools import product


@dataclass(frozen=True, eq=True)
class UserOption:  # 1
    nickname: str
    hobby: str
    email: str

gidgid

from typing import List
from more_itertools import chunked


def batch_insert(col, items: List[dict], batch_size: int):
    batches = chunked(items, batch_size)  # 1
    for batch in batches:
        col.insert_many(batch)  # 2