Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More
How to use get_macro_function method in dbt-osmosis
Best Python code snippet using dbt-osmosis_python
lmapi.py
Source:lmapi.py 
...614# macro functions615 @log_d616 @api_need_master617 @notify_plugins618 def get_macro_function(self, **kwargs):619 i = parse_api_params(kwargs, 'i', 'S')620 f = eva.lm.controller.get_macro_function(i)621 if not f:622 raise ResourceNotFound623 return f624 @log_d625 @api_need_master626 @notify_plugins627 def list_macro_functions(self, **kwargs):628 fn = eva.lm.controller.get_macro_function()629 result = []630 for f, v in fn.items():631 v = v.copy()632 del v['src']633 result.append(v)634 return sorted(result, key=lambda k: k['name'])635 @log_w636 @api_need_master637 @notify_plugins638 def put_macro_function(self, **kwargs):639 name, description, i, o, code = parse_api_params(640 kwargs, ['function', 'description', 'input', 'output', 'src'],641 'ss...')642 fname = eva.lm.controller.put_macro_function(fname=name,643 fdescr=description,644 i=i,645 o=o,646 fcode=code)647 if not fname:648 raise FunctionFailed649 return eva.lm.controller.get_macro_function(fname)650 @log_i651 @api_need_master652 @notify_plugins653 def reload_macro_function(self, **kwargs):654 i, tp = parse_api_params(kwargs, 'ip', 'ss')655 if not eva.lm.controller.reload_macro_function(fname=i, tp=tp):656 raise FunctionFailed657 else:658 return True659 @log_w660 @api_need_master661 @notify_plugins662 def destroy_macro_function(self, **kwargs):663 i = parse_api_params(kwargs, 'i', 'S')664 return eva.lm.controller.destroy_macro_function(i)665# macros666 @log_d667 @notify_plugins668 def groups_macro(self, **kwargs):669 """670 get macro groups list671 Get the list of macros. Useful e.g. for custom interfaces.672 Args:673 k:674 """675 k = parse_function_params(kwargs, 'k', '.')676 result = []677 for i, v in eva.lm.controller.macros_by_id.copy().items():678 if key_check(k, v) and \679 v.group not in result:680 result.append(v.group)681 return sorted(result)682 @log_d683 @notify_plugins684 def list_macros(self, **kwargs):685 """686 get macro list687 Get the list of all available :doc:`macros<macros>`.688 Args:689 k:690 Optional:691 .g: filter by group692 """693 k, group = parse_function_params(kwargs, 'kg', '.s')694 result = []695 for i, v in eva.lm.controller.macros_by_id.copy().items():696 if key_check(k, v) and \697 (not group or eva.item.item_match(v, [], [ group ])):698 result.append(v.serialize(info=True))699 return sorted(result, key=lambda k: k['full_id'])700 @log_i701 @api_need_master702 @notify_plugins703 def create_macro(self, **kwargs):704 """705 create new macro706 Creates new :doc:`macro<macros>`. Macro code should be put in **xc/lm**707 manually.708 Args:709 k: .master710 .i: macro id711 Optional:712 .g: macro group713 """714 k, i, g, save = parse_function_params(kwargs, 'kigS', '.Ssb')715 save = save or eva.core.config.auto_save716 return eva.lm.controller.create_macro(i, g, save).serialize()717 @log_w718 @api_need_master719 @notify_plugins720 def destroy_macro(self, **kwargs):721 """722 delete macro723 Deletes :doc:`macro<macros>`.724 Args:725 k: .master726 .i: macro id727 """728 i = parse_api_params(kwargs, 'i', 'S')729 return eva.lm.controller.destroy_macro(i)730 @log_i731 @api_need_master732 @notify_plugins733 def list_macro_props(self, **kwargs):734 """735 get macro configuration properties736 Args:737 k: .master738 .i: macro id739 """740 i = parse_api_params(kwargs, 'i', 'S')741 item = eva.lm.controller.get_macro(i)742 if not item:743 raise ResourceNotFound744 return item.serialize(props=True)745 @log_i746 @api_need_master747 @notify_plugins748 def set_macro_prop(self, **kwargs):749 """750 set macro configuration property751 Set configuration parameters of the :doc:`macro<macros>`.752 Args:753 k: .master754 .i: item id755 .p: property name (or empty for batch set)756 757 Optional:758 .v: propery value (or dict for batch set)759 save: save configuration after successful call760 """761 i, p, v, save = parse_api_params(kwargs, 'ipvS', 's..b')762 save = save or eva.core.config.auto_save763 if not p and not isinstance(v, dict):764 raise InvalidParameter('property not specified')765 if is_oid(i):766 t, i = parse_oid(i)767 macro = eva.lm.controller.get_macro(i)768 if not macro or (is_oid(i) and macro and macro.macro_type != t):769 raise ResourceNotFound770 return self._set_prop(macro, p, v, save)771 @log_d772 @notify_plugins773 def get_macro(self, **kwargs):774 """775 get macro information776 Args:777 k:778 .i: macro id779 """780 k, i = parse_function_params(kwargs, 'ki', '.S')781 item = eva.lm.controller.get_macro(i)782 if not item or not key_check(k, item):783 raise ResourceNotFound784 result = item.serialize(info=True)785 if key_check_master(k):786 t, s = eva.lm.controller.get_macro_source(item)787 result['src'] = s788 if t:789 result['type'] += ':' + t790 return result791# cycle functions792 @log_d793 @notify_plugins794 def groups_cycle(self, **kwargs):795 """796 get cycle groups list797 Get the list of cycles. Useful e.g. for custom interfaces.798 Args:799 k:800 """801 k = parse_function_params(kwargs, 'k', '.')802 result = []803 for i, v in eva.lm.controller.cycles_by_id.copy().items():804 if key_check(k, v, ro_op=True) and \805 v.group not in result:806 result.append(v.group)807 return sorted(result)808 @log_d809 @notify_plugins810 def list_cycles(self, **kwargs):811 """812 get cycle list813 Get the list of all available :doc:`cycles<cycles>`.814 Args:815 k:816 Optional:817 .g: filter by group818 """819 k, group = parse_function_params(kwargs, 'kg', '.s')820 result = []821 for i, v in eva.lm.controller.cycles_by_id.copy().items():822 if key_check(k, v, ro_op=True) and \823 (not group or eva.item.item_match(v, [], [ group ])):824 result.append(v.serialize(info=True))825 return sorted(result, key=lambda k: k['full_id'])826 @log_i827 @api_need_master828 @notify_plugins829 def create_cycle(self, **kwargs):830 """831 create new cycle832 Creates new :doc:`cycle<cycles>`.833 Args:834 k: .master835 .i: cycle id836 Optional:837 .g: cycle group838 .v: cycle properties (dict) or human-readable input839 """840 k, i, g, v, save = parse_function_params(kwargs, 'kigvS', '.Ss.b')841 save = save or eva.core.config.auto_save842 cycle = eva.lm.controller.create_cycle(i, g, save)843 try:844 if v:845 if isinstance(v, dict):846 self._set_prop(cycle, v=v, save=save)847 else:848 cycle.set_hri(v, save=save)849 except:850 eva.core.log_traceback()851 eva.lm.controller.destroy_cycle(cycle.item_id)852 raise853 return cycle.serialize(info=True)854 @log_w855 @api_need_master856 @notify_plugins857 def destroy_cycle(self, **kwargs):858 """859 delete cycle860 Deletes :doc:`cycle<cycles>`. If cycle is running, it is stopped before861 deletion.862 Args:863 k: .master864 .i: cycle id865 """866 i = parse_api_params(kwargs, 'i', 'S')867 return eva.lm.controller.destroy_cycle(i)868 @log_i869 @api_need_master870 @notify_plugins871 def list_cycle_props(self, **kwargs):872 """873 get cycle configuration properties874 Args:875 k: .master876 .i: cycle id877 """878 i = parse_api_params(kwargs, 'i', 'S')879 item = eva.lm.controller.get_cycle(i)880 if not item:881 raise ResourceNotFound882 return item.serialize(props=True)883 @log_i884 @api_need_master885 @notify_plugins886 def set_cycle_prop(self, **kwargs):887 """888 set cycle property889 Set configuration parameters of the :doc:`cycle<cycles>`.890 Args:891 k: .master892 .i: item id893 .p: property name (or empty for batch set)894 895 Optional:896 .v: propery value (or dict for batch set)897 save: save configuration after successful call898 """899 i, p, v, save = parse_api_params(kwargs, 'ipvS', 's..b')900 save = save or eva.core.config.auto_save901 if not p and not isinstance(v, dict):902 raise InvalidParameter('property not specified')903 if is_oid(i):904 t, i = parse_oid(i)905 cycle = eva.lm.controller.get_cycle(i)906 if not cycle or (is_oid(i) and cycle and cycle.cycle_type != t):907 raise ResourceNotFound908 return self._set_prop(cycle, p, v, save)909 @log_d910 @notify_plugins911 def get_cycle(self, **kwargs):912 """913 get cycle information914 Args:915 k:916 .i: cycle id917 Returns:918 field "value" contains real average cycle interval919 """920 k, i = parse_function_params(kwargs, 'ki', '.S')921 item = eva.lm.controller.get_cycle(i)922 if not item or not key_check(k, item, ro_op=True):923 raise ResourceNotFound924 return item.serialize(info=True)925 @log_i926 @notify_plugins927 def start_cycle(self, **kwargs):928 """929 start cycle930 Args:931 k:932 .i: cycle id933 """934 k, i = parse_function_params(kwargs, 'ki', '.S')935 cycle = eva.lm.controller.get_cycle(i)936 if not cycle:937 raise ResourceNotFound938 elif not key_check(k, cycle):939 raise AccessDenied940 if cycle.cycle_status:941 raise ResourceBusy('cycle is already started')942 return cycle.start()943 @log_i944 @notify_plugins945 def stop_cycle(self, **kwargs):946 """947 stop cycle948 Args:949 k:950 .i: cycle id951 Optional:952 wait: wait until cycle is stopped953 """954 k, i, wait = parse_function_params(kwargs, 'kiW', '.Sb')955 cycle = eva.lm.controller.get_cycle(i)956 if not cycle:957 raise ResourceNotFound958 elif not key_check(k, cycle):959 raise AccessDenied960 cycle.stop(wait=wait)961 return (True, api_result_accepted) if not wait else True962 @log_i963 @notify_plugins964 def reset_cycle_stats(self, **kwargs):965 """966 reset cycle statistic967 Args:968 k:969 .i: cycle id970 """971 k, i = parse_function_params(kwargs, 'ki', '.S')972 cycle = eva.lm.controller.get_cycle(i)973 if not cycle:974 raise ResourceNotFound975 elif not key_check(k, cycle):976 raise AccessDenied977 cycle.reset_stats()978 return True979# lvars980 @log_i981 @api_need_master982 @notify_plugins983 def get_config(self, **kwargs):984 """985 get lvar configuration986 Args:987 k: .master988 .i: lvaar id989 Returns:990 complete :ref:`lvar<eva3_lvar>` configuration.991 """992 i = parse_api_params(kwargs, 'i', 's')993 if is_oid(i):994 t, i = parse_oid(i)995 item = eva.lm.controller.get_item(i)996 if not item or (is_oid(i) and item and item.item_type != t):997 raise ResourceNotFound998 return item.serialize(config=True)999 @log_i1000 @api_need_master1001 @notify_plugins1002 def save_config(self, **kwargs):1003 """1004 save lvar configuration1005 Saves :ref:`lvar<eva3_lvar>`. configuration on disk (even if it hasn't been1006 changed)1007 Args:1008 k: .master1009 .i: lvar id1010 """1011 i = parse_api_params(kwargs, 'i', 's')1012 if is_oid(i):1013 t, i = parse_oid(i)1014 item = eva.lm.controller.get_item(i)1015 if not item or (is_oid(i) and item and item.item_type != t):1016 raise ResourceNotFound1017 item = eva.lm.controller.get_item(i)1018 return item.save()1019 @log_i1020 @api_need_master1021 @notify_plugins1022 def list(self, **kwargs):1023 """1024 list lvars1025 Args:1026 k: .master1027 Optional:1028 .g: filter by item group1029 x: serialize specified item prop(s)1030 Returns:1031 the list of all :ref:`lvars<eva3_lvar>` available1032 """1033 tp, group, prop = parse_api_params(kwargs, 'pgx', 'ss.', {'p': 'lvar'})1034 if prop:1035 if isinstance(prop, list):1036 pass1037 elif isinstance(prop, str):1038 prop = prop.split(',')1039 else:1040 raise InvalidParameter('"x" must be list or string')1041 result = []1042 if tp == 'LV' or tp == 'lvar':1043 items = eva.lm.controller.lvars_by_full_id1044 else:1045 items = eva.lm.controller.items_by_full_id1046 for i, v in items.copy().items():1047 if not group or eva.item.item_match(v, [], [group]):1048 if not prop:1049 result.append(v.serialize(info=True))1050 else:1051 r = {'oid': v.oid}1052 s = v.serialize(props=True)1053 for p in prop:1054 try:1055 r[p] = s[p]1056 except:1057 raise ResourceNotFound('{}: config prop {}'.format(1058 v.oid, p))1059 result.append(r)1060 result = sorted(result, key=lambda k: k['oid'])1061 if prop:1062 for s in reversed(prop):1063 try:1064 result = sorted(result, key=lambda k: k[s])1065 except:1066 pass1067 return result1068 @log_i1069 @api_need_master1070 @notify_plugins1071 def list_props(self, **kwargs):1072 """1073 list lvar properties1074 Get all editable parameters of the :ref:`lvar<eva3_lvar>` confiugration.1075 Args:1076 k: .master1077 .i: item id1078 """1079 i = parse_api_params(kwargs, 'i', 's')1080 if is_oid(i):1081 t, i = parse_oid(i)1082 item = eva.lm.controller.get_item(i)1083 if not item or (is_oid(i) and item and item.item_type != t):1084 raise ResourceNotFound1085 return item.serialize(props=True)1086 @log_i1087 @api_need_master1088 @notify_plugins1089 def set_prop(self, **kwargs):1090 """1091 set lvar property1092 Set configuration parameters of the :ref:`lvar<eva3_lvar>`.1093 Args:1094 k: .master1095 .i: item id1096 .p: property name (or empty for batch set)1097 1098 Optional:1099 .v: propery value (or dict for batch set)1100 save: save configuration after successful call1101 """1102 i, p, v, save = parse_api_params(kwargs, 'ipvS', 's..b')1103 save = save or eva.core.config.auto_save1104 if not p and not isinstance(v, dict):1105 raise InvalidParameter('property not specified')1106 if is_oid(i):1107 t, i = parse_oid(i)1108 item = eva.lm.controller.get_item(i)1109 if not item or (is_oid(i) and item and item.item_type != t):1110 raise ResourceNotFound1111 return self._set_prop(item, p, v, save)1112 @log_i1113 @api_need_master1114 @notify_plugins1115 def create_lvar(self, **kwargs):1116 """1117 create lvar1118 Create new :ref:`lvar<eva3_lvar>`1119 Args:1120 k: .master1121 .i: lvar id1122 Optional:1123 .g: lvar group1124 save: save lvar configuration immediately1125 """1126 i, g, save = parse_api_params(kwargs, 'igS', 'Ssb')1127 save = save or eva.core.config.auto_save1128 return eva.lm.controller.create_lvar(lvar_id=oid_to_id(i, 'lvar'),1129 group=g,1130 save=save).serialize()1131 @log_i1132 @api_need_master1133 @notify_plugins1134 def create(self, **kwargs):1135 """1136 alias for create_lvar1137 """1138 return self.create_lvar(**kwargs)1139 @log_w1140 @api_need_master1141 @notify_plugins1142 def destroy_lvar(self, **kwargs):1143 """1144 delete lvar1145 Args:1146 k: .master1147 .i: lvar id1148 """1149 i, g = parse_api_params(kwargs, 'ig', 'ss')1150 if not i and not g:1151 raise InvalidParameter('either lvar id or group must be specified')1152 return eva.lm.controller.destroy_item(i) if i \1153 else eva.lm.controller.destroy_group(g)1154 @log_w1155 @api_need_master1156 @notify_plugins1157 def destroy(self, **kwargs):1158 """1159 alias for destroy_lvar1160 """1161 return self.destroy_lvar(**kwargs)1162# controller management1163 @log_i1164 @api_need_master1165 @notify_plugins1166 def list_remote(self, **kwargs):1167 """1168 get a list of items from connected UCs1169 Get a list of the items loaded from the connected :ref:`UC1170 controllers<eva3_lm_remote_uc>`. Useful to debug the controller1171 connections.1172 Args:1173 k: .master1174 Optional:1175 .i: controller id1176 g: filter by item group1177 p: filter by item type1178 """1179 i, group, tp = parse_api_params(kwargs, 'igp', 'sss')1180 result = []1181 items = []1182 if i:1183 controller = eva.lm.controller.get_controller(i)1184 if not controller:1185 raise ResourceNotFound('controller {}'.format(i))1186 c_id = controller.item_id1187 if tp is None or tp in ['U', 'unit', '#']:1188 if i:1189 if not c_id in eva.lm.controller.uc_pool.units_by_controller:1190 return []1191 items.append(1192 eva.lm.controller.uc_pool.units_by_controller[c_id])1193 else:1194 items.append(eva.lm.controller.uc_pool.units_by_controller)1195 if tp is None or tp in ['S', 'sensor', '#']:1196 if i:1197 if not c_id in eva.lm.controller.uc_pool.sensors_by_controller:1198 return []1199 items.append(1200 eva.lm.controller.uc_pool.sensors_by_controller[c_id])1201 else:1202 items.append(eva.lm.controller.uc_pool.sensors_by_controller)1203 if not items:1204 return []1205 if i:1206 for x in items:1207 for a, v in x.copy().items():1208 if not group or eva.item.item_match(v, [], [group]):1209 result.append(v.serialize(full=True))1210 else:1211 for x in items:1212 for c, d in x.copy().items():1213 for a, v in d.copy().items():1214 if not group or eva.item.item_match(v, [], [group]):1215 result.append(v.serialize(full=True))1216 return sorted(sorted(result, key=lambda k: k['oid']),1217 key=lambda k: ['controller_id'])1218 @log_i1219 @api_need_master1220 @notify_plugins1221 def list_controllers(self, **kwargs):1222 """1223 get controllers list1224 Get the list of all connected :ref:`UC controllers<eva3_lm_remote_uc>`.1225 Args:1226 k: .master1227 """1228 result = []1229 for i, v in eva.lm.controller.remote_ucs.copy().items():1230 result.append(v.serialize(info=True))1231 return sorted(result, key=lambda k: k['full_id'])1232 @log_i1233 @api_need_master1234 @notify_plugins1235 def append_controller(self, **kwargs):1236 """1237 connect remote UC via HTTP1238 Connects remote :ref:`UC controller<eva3_lm_remote_uc>` to the local.1239 Args:1240 k: .master1241 u: :doc:`../uc/uc_api` uri (*proto://host:port*, port not required1242 if default)1243 a: remote controller API key (\$key to use local key)1244 Optional:1245 m: ref:`MQTT notifier<mqtt_>` to exchange item states in real time1246 (default: *eva_1*)1247 s: verify remote SSL certificate or pass invalid1248 t: timeout (seconds) for the remote controller API calls1249 save: save connected controller configuration on the disk1250 immediately after creation1251 """1252 uri, key, mqtt_update, ssl_verify, timeout, save = parse_api_params(1253 kwargs, 'uamstS', 'Sssbnb')1254 save = save or eva.core.config.auto_save1255 c = eva.lm.controller.append_controller(uri=uri,1256 key=key,1257 mqtt_update=mqtt_update,1258 ssl_verify=ssl_verify,1259 timeout=timeout,1260 save=save)1261 if not c:1262 raise FunctionFailed1263 return c.serialize(info=True)1264 @log_i1265 @api_need_master1266 @notify_plugins1267 def reload_controller(self, **kwargs):1268 """1269 reload controller1270 Reloads items from connected UC1271 Args:1272 k: .master1273 .i: controller id1274 """1275 i = parse_api_params(kwargs, 'i', 'S')1276 controller = eva.lm.controller.get_controller(i)1277 return eva.lm.controller.uc_pool.manually_reload_controller(1278 controller.item_id)1279 @log_i1280 @api_need_master1281 @notify_plugins1282 def upnp_rescan_controllers(self, **kwargs):1283 """1284 rescan controllers via UPnP1285 Args:1286 k: .master1287 """1288 parse_api_params(kwargs, '', '')1289 import eva.upnp1290 if eva.upnp.discovery_worker.is_active():1291 eva.upnp.discovery_worker.trigger_threadsafe()1292 return True1293 else:1294 return False1295# master functions for lmacro extension management1296 @log_i1297 @api_need_master1298 @notify_plugins1299 def load_ext(self, **kwargs):1300 """1301 load extension module1302 Loads:doc:`macro extension<../lm/ext>`.1303 Args:1304 k: .master1305 .i: extension ID1306 m: extension module1307 Optional:1308 c: extension configuration1309 save: save extension configuration after successful call1310 """1311 i, m, c, save = parse_api_params(kwargs, 'imcS', 'SS.b')1312 save = save or eva.core.config.auto_save1313 if isinstance(c, str):1314 try:1315 c = dict_from_str(c)1316 except:1317 raise InvalidParameter('Unable to parse config')1318 if eva.lm.extapi.load_ext(i, m, c):1319 if save:1320 eva.lm.extapi.save()1321 return eva.lm.extapi.get_ext(i).serialize(full=True, config=True)1322 @log_w1323 @api_need_master1324 @notify_plugins1325 def unload_ext(self, **kwargs):1326 """1327 unload macro extension1328 Args:1329 k: .master1330 .i: extension ID1331 """1332 i = parse_api_params(kwargs, 'i', 'S')1333 eva.lm.extapi.unload_ext(i, remove_data=True)1334 if eva.core.config.auto_save:1335 eva.lm.extapi.save()1336 return True1337 @log_d1338 @api_need_master1339 @notify_plugins1340 def list_ext(self, **kwargs):1341 """1342 get list of available macro extensions1343 Args:1344 k: .master1345 Optional:1346 .full: get full information1347 """1348 full = parse_api_params(kwargs, 'Y', 'b')1349 return sorted(eva.lm.extapi.serialize(full=full, config=full),1350 key=lambda k: k['id'])1351 @log_d1352 @api_need_master1353 @notify_plugins1354 def get_ext(self, **kwargs):1355 """1356 get loaded extension information1357 Args:1358 k: .master1359 .i: extension ID1360 """1361 i = parse_api_params(kwargs, 'i', 'S')1362 ext = eva.lm.extapi.get_ext(i)1363 if ext:1364 return ext.serialize(full=True, config=True)1365 else:1366 raise ResourceNotFound1367 @log_d1368 @api_need_master1369 @notify_plugins1370 def list_ext_mods(self, **kwargs):1371 """1372 get list of available extension modules1373 Args:1374 k: .master1375 """1376 return eva.lm.extapi.list_mods()1377 @log_d1378 @api_need_master1379 @notify_plugins1380 def modinfo_ext(self, **kwargs):1381 """1382 get extension module info1383 Args:1384 k: .master1385 .m: extension module name (without *.py* extension)1386 """1387 m = parse_api_params(kwargs, 'm', 'S')1388 return eva.lm.extapi.modinfo(m)1389 @log_d1390 @api_need_master1391 @notify_plugins1392 def modhelp_ext(self, **kwargs):1393 """1394 get extension usage help1395 Args:1396 k: .master1397 .m: extension name (without *.py* extension)1398 .c: help context (*cfg* or *functions*)1399 """1400 m, c = parse_api_params(kwargs, 'mc', 'SS')1401 return eva.lm.extapi.modhelp(m, c)1402 @log_i1403 @api_need_master1404 @notify_plugins1405 def set_ext_prop(self, **kwargs):1406 """1407 set extension configuration property1408 appends property to extension configuration and reloads module1409 Args:1410 k: .master1411 .i: extension id1412 .p: property name (or empty for batch set)1413 Optional:1414 .v: propery value (or dict for batch set)1415 save: save configuration after successful call1416 """1417 i, p, v, save = parse_api_params(kwargs, 'ipvS', 'S.Rb')1418 save = save or eva.core.config.auto_save1419 eva.lm.extapi.set_ext_prop(i, p, v)1420 if save:1421 eva.lm.extapi.save()1422 return True1423class LM_HTTP_API_abstract(LM_API, GenericHTTP_API):1424 def __init__(self):1425 super().__init__()1426 self._nofp_log('put_macro_function', 'src')1427 self._nofp_log('put_macro_function', 'input')1428 self._nofp_log('put_macro_function', 'output')1429 self._nofp_log('set_macro_prop', 'v')1430class LM_HTTP_API(LM_HTTP_API_abstract, GenericHTTP_API):1431 def __init__(self):1432 super().__init__()1433 self.expose_api_methods('lmapi')1434 self.wrap_exposed()1435class LM_JSONRPC_API(eva.sysapi.SysHTTP_API_abstract,1436 eva.api.JSON_RPC_API_abstract, LM_HTTP_API_abstract):1437 def __init__(self):1438 super().__init__()1439 self.expose_api_methods('lmapi', set_api_uri=False)1440 self.expose_api_methods('sysapi', set_api_uri=False)1441class LM_REST_API(eva.sysapi.SysHTTP_API_abstract,1442 eva.sysapi.SysHTTP_API_REST_abstract,1443 eva.api.GenericHTTP_API_REST_abstract, LM_HTTP_API_abstract,1444 GenericHTTP_API):1445 @generic_web_api_method1446 @restful_api_method1447 def GET(self, rtp, k, ii, save, kind, method, for_dir, props):1448 try:1449 return super().GET(rtp, k, ii, save, kind, method, for_dir, props)1450 except MethodNotFound:1451 pass1452 if rtp == 'lvar':1453 if kind == 'groups':1454 return self.groups(k=k, p=rtp)1455 elif kind == 'history':1456 return self.state_history(k=k, i=ii, **props)1457 elif kind == 'log':1458 return self.state_log(k=k, i='{}:{}'.format(rtp, ii), **props)1459 elif kind == 'props':1460 return self.list_props(k=k, i=ii)1461 elif kind == 'config':1462 return self.get_config(k=k, i=ii)1463 elif for_dir:1464 return self.state(k=k, p=rtp, g=ii, **props)1465 else:1466 return self.state(k=k, p=rtp, i=ii, **props)1467 elif rtp == 'controller':1468 if kind == 'items':1469 return self.list_remote(k=k, i=ii, **props)1470 elif kind == 'props' and ii and ii.find('/') != -1:1471 return self.list_controller_props(k=k, i=ii)1472 else:1473 if ii and ii.find('/') != -1:1474 return self.get_controller(k=k, i=ii)1475 else:1476 return self.list_controllers(k=k, g=ii)1477 elif rtp == 'action':1478 return self.result(k=k, u=ii, **props)1479 elif rtp == 'dmatrix_rule':1480 if ii:1481 if kind == 'props':1482 return self.list_rule_props(k=k, i=ii)1483 else:1484 return self.get_rule(k=k, i=ii)1485 else:1486 return self.list_rules(k=k)1487 elif rtp == 'job':1488 if ii:1489 if kind == 'props':1490 return self.list_job_props(k=k, i=ii)1491 else:1492 return self.get_job(k=k, i=ii)1493 else:1494 return self.list_jobs(k=k)1495 elif rtp == 'lmacro':1496 if ii:1497 if kind == 'props':1498 return self.list_macro_props(k=k, i=ii)1499 else:1500 if for_dir:1501 return self.list_macros(k=k, g=ii)1502 else:1503 return self.get_macro(k=k, i=ii)1504 else:1505 if kind == 'groups':1506 return self.groups_macro(k=k)1507 else:1508 return self.list_macros(k=k)1509 elif rtp == 'lmacro-function':1510 if ii:1511 return self.get_macro_function(k=k, i=ii)1512 else:1513 return self.list_macro_functions(k=k)1514 elif rtp == 'lcycle':1515 if ii:1516 if kind == 'props':1517 return self.list_cycle_props(k=k, i=ii)1518 else:1519 if for_dir:1520 return self.list_cycles(k=k, g=ii)1521 else:1522 return self.get_cycle(k=k, i=ii)1523 else:1524 if kind == 'groups':1525 return self.groups_cycle(k=k)...
osmosis.py
Source:osmosis.py
...375 if node.original_file_path == path:376 return node377 return None378 @lru_cache(maxsize=100)379 def get_macro_function(self, macro_name: str) -> Callable[[Dict[str, Any]], Any]:380 """Get macro as a function which takes a dict via argument named `kwargs`,381 ie: `kwargs={"relation": ...}`382 make_schema_fn = get_macro_function('make_schema')\n383 make_schema_fn({'name': '__test_schema_1'})\n384 make_schema_fn({'name': '__test_schema_2'})"""385 return partial(self.adapter.execute_macro, macro_name=macro_name, manifest=self.dbt)386 def adapter_execute(387 self, sql: str, auto_begin: bool = False, fetch: bool = False388 ) -> Tuple[AdapterResponse, agate.Table]:389 """Wraps adapter.execute. Execute SQL against database"""390 return self.adapter.execute(sql, auto_begin, fetch)391 def execute_macro(392 self,393 macro: str,394 kwargs: Optional[Dict[str, Any]] = None,395 ) -> Any:396 """Wraps adapter execute_macro. Execute a macro like a function."""397 return self.get_macro_function(macro)(kwargs=kwargs)398 def execute_sql(self, raw_sql: str) -> DbtAdapterExecutionResult:399 """Execute dbt SQL statement against database"""400 # if no jinja chars then these are synonymous401 compiled_sql = raw_sql402 if has_jinja(raw_sql):403 # jinja found, compile it404 compiled_sql = self.compile_sql(raw_sql).compiled_sql405 return DbtAdapterExecutionResult(406 *self.adapter_execute(compiled_sql, fetch=True),407 raw_sql,408 compiled_sql,409 )410 def execute_node(self, node: ManifestNode) -> DbtAdapterExecutionResult:411 """Execute dbt SQL statement against database from a ManifestNode"""...
controller.py
Source:controller.py
...663 except:664 eva.core.log_traceback()665 return False666 return True667def get_macro_function(fname=None):668 return eva.lm.plc.get_macro_function(fname)669def get_macro_source(macro_id):670 if isinstance(macro_id, str):671 macro = get_macro(macro_id)672 else:673 macro = macro_id674 if not macro:675 return None676 file_name = eva.core.format_xc_fname(677 fname=macro.action_exec if macro.action_exec else '{}.py'.678 format(macro.item_id))679 if os.path.isfile(file_name):680 with open(file_name) as fd:681 code = fd.read()682 if code.startswith('# SFC'):...
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.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run dbt-osmosis automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
