How to use update_from method in localstack

Best Python code snippet using localstack_python

make_profiles.py

Source:make_profiles.py

...51            if user_id in target_users:52                user = user_counters[user_id]53                user.add(OT_GLOBAL, CT_TRANSACTION, RT_SUM, '', 1, ts)54                user.add(OT_ITEM, CT_TRANSACTION, RT_SUM, item_id, 1, ts)55                user.update_from(item, OT_ATTR, CT_HAS, RT_SUM, CT_TRANSACTION, RT_SUM, ts)56                user.update_from(item, OT_AVAIL, CT_HAS, RT_SUM, CT_TRANSACTION, RT_SUM, ts)57                user.update_from(item, OT_DURATION, CT_VALUE, RT_SUM, CT_TRANSACTION, RT_SUM, ts)58                user.update_from(item, OT_FEATURE, CT_VALUE, RT_SUM, CT_TRANSACTION, RT_SUM, ts)59                user.update_from(item, OT_ITEMTYPE, CT_HAS, RT_SUM, CT_TRANSACTION, RT_SUM, ts)60                user.add(OT_WATCHTIME, CT_TRANSACTION, RT_SUM, '', watch_time, ts)61                user.add(OT_DEVICETYPE, CT_TRANSACTION, RT_SUM, d_type, 1, ts)62                user.add(OT_DEVICEMAN, CT_TRANSACTION, RT_SUM, d_man, 1, ts)63                user.add(OT_WATCHMODE, CT_TRANSACTION, RT_SUM, mode, 1, ts)64            item.add(OT_WATCHTIME, CT_TRANSACTION, RT_SUM, '', watch_time, ts)65            item.add(OT_DEVICETYPE, CT_TRANSACTION, RT_SUM, d_type, 1, ts)66            item.add(OT_DEVICEMAN, CT_TRANSACTION, RT_SUM, d_man, 1, ts)67            item.add(OT_WATCHMODE, CT_TRANSACTION, RT_SUM, mode, 1, ts)68    print "Parsing bookmarks"69    with open(sys.argv[3]) as csvfile:70        reader = csv.reader(csvfile, delimiter=',')71        count = 172        for row in reader:73            if count % 100000 == 0:74                print count75            count += 176            user_id = int(row[0])77            item_id = int(row[1])78            ts = int(float(row[2]) * 6)79            item = item_counters[item_id]80            item.add(OT_GLOBAL, CT_BOOKMARK, RT_SUM, '', 1, ts)81            item.add(OT_GLOBAL, CT_BOOKMARK, RT_7D, '', 1, ts)82            item.add(OT_GLOBAL, CT_BOOKMARK, RT_30D, '', 1, ts)83            if user_id in target_users:84                user = user_counters[user_id]85                user.add(OT_GLOBAL, CT_BOOKMARK, RT_SUM, '', 1, ts)86                user.add(OT_ITEM, CT_BOOKMARK, RT_SUM, item_id, 1, ts)87                user.update_from(item, OT_ATTR, CT_HAS, RT_SUM, CT_BOOKMARK, RT_SUM, ts)88                user.update_from(item, OT_AVAIL, CT_HAS, RT_SUM, CT_BOOKMARK, RT_SUM, ts)89                user.update_from(item, OT_DURATION, CT_VALUE, RT_SUM, CT_BOOKMARK, RT_SUM, ts)90                user.update_from(item, OT_FEATURE, CT_VALUE, RT_SUM, CT_BOOKMARK, RT_SUM, ts)91                user.update_from(item, OT_ITEMTYPE, CT_HAS, RT_SUM, CT_BOOKMARK, RT_SUM, ts)92    print "Parsing ratings"93    with open(sys.argv[4]) as csvfile:94        reader = csv.reader(csvfile, delimiter=',')95        count = 196        for row in reader:97            if count % 100000 == 1:98                print count99            count += 1100            user_id = int(row[0])101            item_id = int(row[1])102            rating = int(row[2])103            ts = int(float(row[3]) * 6)104            item = item_counters[item_id]105            item.add(OT_GLOBAL, CT_HAS_RATING, RT_SUM, '', 1, ts)106            item.add(OT_GLOBAL, CT_HAS_RATING, RT_7D, '', 1, ts)107            item.add(OT_GLOBAL, CT_HAS_RATING, RT_30D, '', 1, ts)108            item.add(OT_GLOBAL, CT_RATING, RT_SUM, '', rating, ts)109            if user_id in target_users:110                user = user_counters[user_id]111                user.add(OT_GLOBAL, CT_HAS_RATING, RT_SUM, '', 1, ts)112                user.add(OT_ITEM, CT_RATING, RT_SUM, item_id, rating, ts)113                user.update_from(item, OT_ATTR, CT_HAS, RT_SUM, CT_HAS_RATING, RT_SUM, ts)114                user.update_from(item, OT_AVAIL, CT_HAS, RT_SUM, CT_HAS_RATING, RT_SUM, ts)115                user.update_from(item, OT_DURATION, CT_VALUE, RT_SUM, CT_HAS_RATING, RT_SUM, ts)116                user.update_from(item, OT_FEATURE, CT_VALUE, RT_SUM, CT_HAS_RATING, RT_SUM, ts)117                user.update_from(item, OT_ITEMTYPE, CT_HAS, RT_SUM, CT_HAS_RATING, RT_SUM, ts)118                user.update_from(item, OT_ATTR, CT_HAS, RT_SUM, CT_RATING, RT_SUM, rating, ts)119                user.update_from(item, OT_AVAIL, CT_HAS, RT_SUM, CT_RATING, RT_SUM, rating, ts)120                user.update_from(item, OT_DURATION, CT_VALUE, RT_SUM, CT_RATING, RT_SUM, rating, ts)121                user.update_from(item, OT_FEATURE, CT_VALUE, RT_SUM, CT_RATING, RT_SUM, rating, ts)122                user.update_from(item, OT_ITEMTYPE, CT_HAS, RT_SUM, CT_RATING, RT_SUM, rating, ts)123    print "Dumping user profiles"124    with open(sys.argv[6], 'w') as user_pickle:125        pickle.dump(user_counters, user_pickle)126    print "Dumping item profiles"127    with open(sys.argv[7], 'w') as item_pickle:...

get_settings.py

Source:get_settings.py

1from typing import *2from ..utils.settings import get_setting3from ..typings import *4__all__ = [5    "extract_kwargs_for_field",6    "get_default_kwargs",7    "get_kwargs",8    "extract_model_kwargs"9]10_COMMON_DEFAULT_KWARGS_UPDATE = {11    "max_length": 127,12}13def extract_kwargs_for_field(14        kwargs: Kwargs,15        field: str,16        update_from: Optional[Kwargs] = None,17) -> Kwargs:18    if update_from is None:19        update_from = {}20    else:21        update_from = update_from.copy()22    23    new_kwargs = kwargs.get(field, kwargs)24    update_from.update(new_kwargs)25    26    return update_from27def get_default_kwargs() -> Kwargs:28    """Returns the default kwargs used for all fields"""29    default_kwargs: Kwargs = get_setting("COMMON_DEFAULT_KWARGS", {})30    default_kwargs.update(_COMMON_DEFAULT_KWARGS_UPDATE)31    32    return default_kwargs33def get_kwargs(class_name: str) -> Kwargs:34    """Returns the specific kwargs for each field `class_name`"""35    default_kwargs = get_default_kwargs()36    class_kwargs = get_setting("COMMON_KWARGS", {})37    38    use_kwargs = class_kwargs.get(class_name, default_kwargs)39    40    return use_kwargs41def extract_model_kwargs(class_name: str, field: str, update_from: Optional[Kwargs] = None) -> Kwargs:42    kwargs: Kwargs = get_kwargs(class_name)43    44    return extract_kwargs_for_field(kwargs.get(45        field, get_default_kwargs()...

loader.py

Source:loader.py

1import requests2from django.conf import settings3from synchroniser.static.static_values import StaticSyncValues4class RequestList:5    @staticmethod6    def load_one_row_by_url(uri, params=None):7        return RequestList.__load(uri, 1, 1, update_from=None, params=params)8    @staticmethod9    def load_rows_by_url(uri, limit=StaticSyncValues.load_limit, updated_from=None, params=None):10        result = []11        offset = 012        while offset < 10000:13            rows = RequestList.__load(uri, limit, offset, updated_from, params)['rows']14            if not rows:15                break16            result.extend(rows)17            offset += limit18        return result19    @staticmethod20    def __load(uri, limit, offset, update_from=None, params=None):21        payload = {'limit': limit, 'offset': offset, 'archived': 'All'}22        if update_from:23            payload['update_from'] = update_from24        if params:25            payload.update(params)26        headers = {'Content-Type': 'application/json', 'Authorization': 'Basic ' + settings.MOYSKLAD_TOKEN}27        return requests.get(uri, params=payload, headers=headers).json()28    @staticmethod29    def upload(url, json_data):30        headers = {'Content-Type': 'application/json', 'Authorization': 'Basic ' + settings.MOYSKLAD_TOKEN}...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.