KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use enable_gzip method in Playwright Python

Best Python code snippet using playwright-python

_base.py

Source:_base.py

1"""Commons function for Sync and Async client."""2from __future__ import absolute_import3import base644import codecs5import configparser6import csv7import os8from datetime import datetime, timedelta9from typing import Iterator, List, Generator, Any, Union, Iterable, AsyncGenerator10from urllib3 import HTTPResponse11from influxdb_client import Configuration, Dialect, Query, OptionStatement, VariableAssignment, Identifier, \12    Expression, BooleanLiteral, IntegerLiteral, FloatLiteral, DateTimeLiteral, UnaryExpression, DurationLiteral, \13    Duration, StringLiteral, ArrayExpression, ImportDeclaration, MemberExpression, MemberAssignment, File, \14    WriteService, QueryService, DeleteService, DeletePredicateRequest15from influxdb_client.client.flux_csv_parser import FluxResponseMetadataMode, FluxCsvParser, FluxSerializationMode16from influxdb_client.client.flux_table import FluxTable, FluxRecord17from influxdb_client.client.util.date_utils import get_date_helper18from influxdb_client.client.util.helpers import get_org_query_param19from influxdb_client.client.write.dataframe_serializer import DataframeSerializer20from influxdb_client.rest import _UTF_8_encoding21try:22    import dataclasses23    _HAS_DATACLASS = True24except ModuleNotFoundError:25    _HAS_DATACLASS = False26# noinspection PyMethodMayBeStatic27class _BaseClient(object):28    def __init__(self, url, token, debug=None, timeout=10_000, enable_gzip=False, org: str = None,29                 default_tags: dict = None, **kwargs) -> None:30        self.url = url31        self.token = token32        self.org = org33        self.default_tags = default_tags34        self.conf = _Configuration()35        if self.url.endswith("/"):36            self.conf.host = self.url[:-1]37        else:38            self.conf.host = self.url39        self.conf.enable_gzip = enable_gzip40        self.conf.debug = debug41        self.conf.verify_ssl = kwargs.get('verify_ssl', True)42        self.conf.ssl_ca_cert = kwargs.get('ssl_ca_cert', None)43        self.conf.proxy = kwargs.get('proxy', None)44        self.conf.proxy_headers = kwargs.get('proxy_headers', None)45        self.conf.connection_pool_maxsize = kwargs.get('connection_pool_maxsize', self.conf.connection_pool_maxsize)46        self.conf.timeout = timeout47        auth_token = self.token48        self.auth_header_name = "Authorization"49        self.auth_header_value = "Token " + auth_token50        auth_basic = kwargs.get('auth_basic', False)51        if auth_basic:52            self.auth_header_value = "Basic " + base64.b64encode(token.encode()).decode()53        self.retries = kwargs.get('retries', False)54        self.profilers = kwargs.get('profilers', None)55        pass56    def _version(self, response) -> str:57        if response is not None and len(response) >= 3:58            if 'X-Influxdb-Version' in response[2]:59                return response[2]['X-Influxdb-Version']60        return "unknown"61    @classmethod62    def _from_config_file(cls, config_file: str = "config.ini", debug=None, enable_gzip=False):63        config = configparser.ConfigParser()64        config.read(config_file)65        def config_value(key: str):66            return config['influx2'][key].strip('"')67        url = config_value('url')68        token = config_value('token')69        timeout = None70        if config.has_option('influx2', 'timeout'):71            timeout = config_value('timeout')72        org = None73        if config.has_option('influx2', 'org'):74            org = config_value('org')75        verify_ssl = True76        if config.has_option('influx2', 'verify_ssl'):77            verify_ssl = config_value('verify_ssl')78        ssl_ca_cert = None79        if config.has_option('influx2', 'ssl_ca_cert'):80            ssl_ca_cert = config_value('ssl_ca_cert')81        connection_pool_maxsize = None82        if config.has_option('influx2', 'connection_pool_maxsize'):83            connection_pool_maxsize = config_value('connection_pool_maxsize')84        auth_basic = False85        if config.has_option('influx2', 'auth_basic'):86            auth_basic = config_value('auth_basic')87        default_tags = None88        if config.has_section('tags'):89            tags = {k: v.strip('"') for k, v in config.items('tags')}90            default_tags = dict(tags)91        profilers = None92        if config.has_option('influx2', 'profilers'):93            profilers = [x.strip() for x in config_value('profilers').split(',')]94        proxy = None95        if config.has_option('influx2', 'proxy'):96            proxy = config_value('proxy')97        return cls(url, token, debug=debug, timeout=_to_int(timeout), org=org, default_tags=default_tags,98                   enable_gzip=enable_gzip, verify_ssl=_to_bool(verify_ssl), ssl_ca_cert=ssl_ca_cert,99                   connection_pool_maxsize=_to_int(connection_pool_maxsize), auth_basic=_to_bool(auth_basic),100                   profilers=profilers, proxy=proxy)101    @classmethod102    def _from_env_properties(cls, debug=None, enable_gzip=False):103        url = os.getenv('INFLUXDB_V2_URL', "http://localhost:8086")104        token = os.getenv('INFLUXDB_V2_TOKEN', "my-token")105        timeout = os.getenv('INFLUXDB_V2_TIMEOUT', "10000")106        org = os.getenv('INFLUXDB_V2_ORG', "my-org")107        verify_ssl = os.getenv('INFLUXDB_V2_VERIFY_SSL', "True")108        ssl_ca_cert = os.getenv('INFLUXDB_V2_SSL_CA_CERT', None)109        connection_pool_maxsize = os.getenv('INFLUXDB_V2_CONNECTION_POOL_MAXSIZE', None)110        auth_basic = os.getenv('INFLUXDB_V2_AUTH_BASIC', "False")111        prof = os.getenv("INFLUXDB_V2_PROFILERS", None)112        profilers = None113        if prof is not None:114            profilers = [x.strip() for x in prof.split(',')]115        default_tags = dict()116        for key, value in os.environ.items():117            if key.startswith("INFLUXDB_V2_TAG_"):118                default_tags[key[16:].lower()] = value119        return cls(url, token, debug=debug, timeout=_to_int(timeout), org=org, default_tags=default_tags,120                   enable_gzip=enable_gzip, verify_ssl=_to_bool(verify_ssl), ssl_ca_cert=ssl_ca_cert,121                   connection_pool_maxsize=_to_int(connection_pool_maxsize), auth_basic=_to_bool(auth_basic),122                   profilers=profilers)123# noinspection PyMethodMayBeStatic124class _BaseQueryApi(object):125    default_dialect = Dialect(header=True, delimiter=",", comment_prefix="#",126                              annotations=["datatype", "group", "default"], date_time_format="RFC3339")127    def __init__(self, influxdb_client, query_options=None):128        from influxdb_client.client.query_api import QueryOptions129        self._query_options = QueryOptions() if query_options is None else query_options130        self._influxdb_client = influxdb_client131        self._query_api = QueryService(influxdb_client.api_client)132    """Base implementation for Queryable API."""133    def _to_tables(self, response, query_options=None, response_metadata_mode:134                   FluxResponseMetadataMode = FluxResponseMetadataMode.full) -> List[FluxTable]:135        """136        Parse HTTP response to FluxTables.137        :param response: HTTP response from an HTTP client. Expected type: `urllib3.response.HTTPResponse`.138        """139        _parser = self._to_tables_parser(response, query_options, response_metadata_mode)140        list(_parser.generator())141        return _parser.table_list()142    async def _to_tables_async(self, response, query_options=None, response_metadata_mode:143                               FluxResponseMetadataMode = FluxResponseMetadataMode.full) -> List[FluxTable]:144        """145        Parse HTTP response to FluxTables.146        :param response: HTTP response from an HTTP client. Expected type: `aiohttp.client_reqrep.ClientResponse`.147        """148        async with self._to_tables_parser(response, query_options, response_metadata_mode) as parser:149            async for _ in parser.generator_async():150                pass151            return parser.table_list()152    def _to_csv(self, response: HTTPResponse) -> Iterator[List[str]]:153        """Parse HTTP response to CSV."""154        return csv.reader(codecs.iterdecode(response, _UTF_8_encoding))155    def _to_flux_record_stream(self, response, query_options=None,156                               response_metadata_mode: FluxResponseMetadataMode = FluxResponseMetadataMode.full) -> \157            Generator[FluxRecord, Any, None]:158        """159        Parse HTTP response to FluxRecord stream.160        :param response: HTTP response from an HTTP client. Expected type: `urllib3.response.HTTPResponse`.161        """162        _parser = self._to_flux_record_stream_parser(query_options, response, response_metadata_mode)163        return _parser.generator()164    async def _to_flux_record_stream_async(self, response, query_options=None, response_metadata_mode:165                                           FluxResponseMetadataMode = FluxResponseMetadataMode.full) -> \166            AsyncGenerator['FluxRecord', None]:167        """168        Parse HTTP response to FluxRecord stream.169        :param response: HTTP response from an HTTP client. Expected type: `aiohttp.client_reqrep.ClientResponse`.170        """171        _parser = self._to_flux_record_stream_parser(query_options, response, response_metadata_mode)172        return (await _parser.__aenter__()).generator_async()173    def _to_data_frame_stream(self, data_frame_index, response, query_options=None,174                              response_metadata_mode: FluxResponseMetadataMode = FluxResponseMetadataMode.full):175        """176        Parse HTTP response to DataFrame stream.177        :param response: HTTP response from an HTTP client. Expected type: `urllib3.response.HTTPResponse`.178        """179        _parser = self._to_data_frame_stream_parser(data_frame_index, query_options, response, response_metadata_mode)180        return _parser.generator()181    async def _to_data_frame_stream_async(self, data_frame_index, response, query_options=None, response_metadata_mode:182                                          FluxResponseMetadataMode = FluxResponseMetadataMode.full):183        """184        Parse HTTP response to DataFrame stream.185        :param response: HTTP response from an HTTP client. Expected type: `aiohttp.client_reqrep.ClientResponse`.186        """187        _parser = self._to_data_frame_stream_parser(data_frame_index, query_options, response, response_metadata_mode)188        return (await _parser.__aenter__()).generator_async()189    def _to_tables_parser(self, response, query_options, response_metadata_mode):190        return FluxCsvParser(response=response, serialization_mode=FluxSerializationMode.tables,191                             query_options=query_options, response_metadata_mode=response_metadata_mode)192    def _to_flux_record_stream_parser(self, query_options, response, response_metadata_mode):193        return FluxCsvParser(response=response, serialization_mode=FluxSerializationMode.stream,194                             query_options=query_options, response_metadata_mode=response_metadata_mode)195    def _to_data_frame_stream_parser(self, data_frame_index, query_options, response, response_metadata_mode):196        return FluxCsvParser(response=response, serialization_mode=FluxSerializationMode.dataFrame,197                             data_frame_index=data_frame_index, query_options=query_options,198                             response_metadata_mode=response_metadata_mode)199    def _to_data_frames(self, _generator):200        """Parse stream of DataFrames into expected type."""201        from ..extras import pd202        if isinstance(_generator, list):203            _dataFrames = _generator204        else:205            _dataFrames = list(_generator)206        if len(_dataFrames) == 0:207            return pd.DataFrame(columns=[], index=None)208        elif len(_dataFrames) == 1:209            return _dataFrames[0]210        else:211            return _dataFrames212    def _org_param(self, org):213        return get_org_query_param(org=org, client=self._influxdb_client)214    def _get_query_options(self):215        if self._query_options and self._query_options.profilers:216            return self._query_options217        elif self._influxdb_client.profilers:218            from influxdb_client.client.query_api import QueryOptions219            return QueryOptions(profilers=self._influxdb_client.profilers)220    def _create_query(self, query, dialect=default_dialect, params: dict = None):221        query_options = self._get_query_options()222        profilers = query_options.profilers if query_options is not None else None223        q = Query(query=query, dialect=dialect, extern=_BaseQueryApi._build_flux_ast(params, profilers))224        if profilers:225            print("\n===============")226            print("Profiler: query")227            print("===============")228            print(query)229        return q230    @staticmethod231    def _params_to_extern_ast(params: dict) -> List['OptionStatement']:232        statements = []233        for key, value in params.items():234            expression = _BaseQueryApi._parm_to_extern_ast(value)235            if expression is None:236                continue237            statements.append(OptionStatement("OptionStatement",238                                              VariableAssignment("VariableAssignment", Identifier("Identifier", key),239                                                                 expression)))240        return statements241    @staticmethod242    def _parm_to_extern_ast(value) -> Union[Expression, None]:243        if value is None:244            return None245        if isinstance(value, bool):246            return BooleanLiteral("BooleanLiteral", value)247        elif isinstance(value, int):248            return IntegerLiteral("IntegerLiteral", str(value))249        elif isinstance(value, float):250            return FloatLiteral("FloatLiteral", value)251        elif isinstance(value, datetime):252            value = get_date_helper().to_utc(value)253            return DateTimeLiteral("DateTimeLiteral", value.strftime('%Y-%m-%dT%H:%M:%S.%fZ'))254        elif isinstance(value, timedelta):255            _micro_delta = int(value / timedelta(microseconds=1))256            if _micro_delta < 0:257                return UnaryExpression("UnaryExpression", argument=DurationLiteral("DurationLiteral", [258                    Duration(magnitude=-_micro_delta, unit="us")]), operator="-")259            else:260                return DurationLiteral("DurationLiteral", [Duration(magnitude=_micro_delta, unit="us")])261        elif isinstance(value, str):262            return StringLiteral("StringLiteral", str(value))263        elif isinstance(value, Iterable):264            return ArrayExpression("ArrayExpression",265                                   elements=list(map(lambda it: _BaseQueryApi._parm_to_extern_ast(it), value)))266        else:267            return value268    @staticmethod269    def _build_flux_ast(params: dict = None, profilers: List[str] = None):270        imports = []271        body = []272        if profilers is not None and len(profilers) > 0:273            imports.append(ImportDeclaration(274                "ImportDeclaration",275                path=StringLiteral("StringLiteral", "profiler")))276            elements = []277            for profiler in profilers:278                elements.append(StringLiteral("StringLiteral", value=profiler))279            member = MemberExpression(280                "MemberExpression",281                object=Identifier("Identifier", "profiler"),282                _property=Identifier("Identifier", "enabledProfilers"))283            prof = OptionStatement(284                "OptionStatement",285                assignment=MemberAssignment(286                    "MemberAssignment",287                    member=member,288                    init=ArrayExpression(289                        "ArrayExpression",290                        elements=elements)))291            body.append(prof)292        if params is not None:293            body.extend(_BaseQueryApi._params_to_extern_ast(params))294        return File(package=None, name=None, type=None, imports=imports, body=body)295class _BaseWriteApi(object):296    def __init__(self, influxdb_client, point_settings=None):297        self._influxdb_client = influxdb_client298        self._point_settings = point_settings299        self._write_service = WriteService(influxdb_client.api_client)300        if influxdb_client.default_tags:301            for key, value in influxdb_client.default_tags.items():302                self._point_settings.add_default_tag(key, value)303    def _append_default_tag(self, key, val, record):304        from influxdb_client import Point305        if isinstance(record, bytes) or isinstance(record, str):306            pass307        elif isinstance(record, Point):308            record.tag(key, val)309        elif isinstance(record, dict):310            record.setdefault("tags", {})311            record.get("tags")[key] = val312        elif isinstance(record, Iterable):313            for item in record:314                self._append_default_tag(key, val, item)315    def _append_default_tags(self, record):316        if self._point_settings.defaultTags and record is not None:317            for key, val in self._point_settings.defaultTags.items():318                self._append_default_tag(key, val, record)319    def _serialize(self, record, write_precision, payload, **kwargs):320        from influxdb_client import Point321        if isinstance(record, bytes):322            payload[write_precision].append(record)323        elif isinstance(record, str):324            self._serialize(record.encode(_UTF_8_encoding), write_precision, payload, **kwargs)325        elif isinstance(record, Point):326            precision_from_point = kwargs.get('precision_from_point', True)327            precision = record.write_precision if precision_from_point else write_precision328            self._serialize(record.to_line_protocol(precision=precision), precision, payload, **kwargs)329        elif isinstance(record, dict):330            self._serialize(Point.from_dict(record, write_precision=write_precision, **kwargs),331                            write_precision, payload, **kwargs)332        elif 'DataFrame' in type(record).__name__:333            serializer = DataframeSerializer(record, self._point_settings, write_precision, **kwargs)334            self._serialize(serializer.serialize(), write_precision, payload, **kwargs)335        elif hasattr(record, "_asdict"):336            # noinspection PyProtectedMember337            self._serialize(record._asdict(), write_precision, payload, **kwargs)338        elif _HAS_DATACLASS and dataclasses.is_dataclass(record):339            self._serialize(dataclasses.asdict(record), write_precision, payload, **kwargs)340        elif isinstance(record, Iterable):341            for item in record:342                self._serialize(item, write_precision, payload, **kwargs)343# noinspection PyMethodMayBeStatic344class _BaseDeleteApi(object):345    def __init__(self, influxdb_client):346        self._influxdb_client = influxdb_client347        self._service = DeleteService(influxdb_client.api_client)348    def _prepare_predicate_request(self, start, stop, predicate):349        date_helper = get_date_helper()350        if isinstance(start, datetime):351            start = date_helper.to_utc(start)352        if isinstance(stop, datetime):353            stop = date_helper.to_utc(stop)354        predicate_request = DeletePredicateRequest(start=start, stop=stop, predicate=predicate)355        return predicate_request356class _Configuration(Configuration):357    def __init__(self):358        Configuration.__init__(self)359        self.enable_gzip = False360    def update_request_header_params(self, path: str, params: dict):361        super().update_request_header_params(path, params)362        if self.enable_gzip:363            # GZIP Request364            if path == '/api/v2/write':365                params["Content-Encoding"] = "gzip"366                params["Accept-Encoding"] = "identity"367                pass368            # GZIP Response369            if path == '/api/v2/query':370                # params["Content-Encoding"] = "gzip"371                params["Accept-Encoding"] = "gzip"372                pass373            pass374        pass375    def update_request_body(self, path: str, body):376        _body = super().update_request_body(path, body)377        if self.enable_gzip:378            # GZIP Request379            if path == '/api/v2/write':380                import gzip381                if isinstance(_body, bytes):382                    return gzip.compress(data=_body)383                else:384                    return gzip.compress(bytes(_body, _UTF_8_encoding))385        return _body386def _to_bool(bool_value):387    return str(bool_value).lower() in ("yes", "true")388def _to_int(int_value):...

pyhotvect.py

Source:pyhotvect.py

1import json2import logging3import os4from datetime import datetime, tzinfo, timedelta5from shutil import copyfile6from typing import Dict, List, Any7import hotvect.mlutils as mlutils8import pandas as pd9from hotvect.utils import trydelete, runshell, read_json, to_zip_archive, ensure_file_exists, clean_dir, \10    ensure_dir_exists11logging.basicConfig(level=logging.WARNING)12# Courtesy of https://stackoverflow.com/a/23705687/23490113# Under CC BY-SA 3.014class SimpleUTC(tzinfo):15    def tzname(self, **kwargs):16        return "UTC"17    def utcoffset(self, dt):18        return timedelta(0)19class Hotvect:20    def __init__(self,21                 hotvect_util_jar_path: str,22                 algorithm_jar_path: str,23                 metadata_base_path: str,24                 output_base_path: str,25                 train_data_path: str,26                 validation_data_path: str,27                 algorithm_definition: Dict[str, Any],28                 state_source_base_path: str = None,29                 run_id: str = "default",30                 enable_gzip: bool = True,31                 jvm_options: List[str] = None32                 ):33        self.algorithm_definition = algorithm_definition34        self.run_id: str = run_id35        self.ran_at: str = datetime.utcnow().replace(tzinfo=SimpleUTC()).isoformat()36        # Utilities37        self.hotvect_util_jar_path = hotvect_util_jar_path38        # Algorithm classes39        self.algorithm_jar_path = algorithm_jar_path40        # Metadata41        self.metadata_base_path = metadata_base_path42        # Output43        self.output_base_path = output_base_path44        # Train data45        self.train_data_path = train_data_path46        ensure_file_exists(train_data_path)47        # Validation data48        self.validation_data_location = validation_data_path49        ensure_file_exists(validation_data_path)50        # State source data51        self.state_source_base_path = state_source_base_path52        if self.state_source_base_path:53            ensure_dir_exists(self.state_source_base_path)54        self.feature_states: Dict[str, str] = {}55        # Gzip56        self.enable_gzip = enable_gzip57        # Jvm options58        if not jvm_options:59            self.jvm_options = ['-Xmx4g']60        else:61            self.jvm_options = jvm_options62        logging.info(f'Initialized:{self.__dict__}')63    def set_run_id(self, run_id: str):64        self.run_id = run_id65    def set_algorithm_definition(self, algorithm_definition: Dict[str, Any]):66        self.algorithm_definition = algorithm_definition67    def metadata_path(self) -> str:68        return os.path.join(self.metadata_base_path, self.algorithm_definition['algorithm_name'], self.run_id)69    def output_path(self) -> str:70        return os.path.join(self.output_base_path, self.algorithm_definition['algorithm_name'], self.run_id)71    def state_output_path(self, state_name: str):72        state_filename = f'{state_name}'73        return os.path.join(74            self.output_path(),75            state_filename76        )77    def encoded_data_file_path(self) -> str:78        encode_suffix = 'encoded.gz' if self.enable_gzip else 'encoded'79        return os.path.join(80            self.output_path(),81            encode_suffix82        )83    def model_file_path(self) -> str:84        return os.path.join(85            self.output_path(),86            'model.parameter'87        )88    def score_file_path(self) -> str:89        return os.path.join(90            self.output_path(),91            'validation_scores.csv'92        )93    def audit_data_file_path(self) -> str:94        encode_suffix = 'audit.jsonl.gz' if self.enable_gzip else 'audit.jsonl'95        return os.path.join(96            self.output_path(),97            encode_suffix98        )99    def predict_parameter_file_path(self) -> str:100        return os.path.join(101            self.output_path(),102            f"{self.algorithm_definition['algorithm_name']}@{self.run_id}.parameters.zip"103        )104    def encode_parameter_file_path(self) -> str:105        return os.path.join(106            self.output_path(),107            'encode.parameters.zip'108        )109    def _write_algorithm_definition(self) -> str:110        """Write algorithm definition so that Java can read it"""111        algorithm_definition_path = os.path.join(112            self.metadata_path(),113            'algorithm_definition.json'114        )115        trydelete(algorithm_definition_path)116        with open(algorithm_definition_path, 'w') as fp:117            json.dump(self.algorithm_definition, fp)118        return algorithm_definition_path119    def _write_data(self, data: Dict, dest_file_name: str) -> str:120        """Write algorithm definition so that Java can read it"""121        dest = os.path.join(122            self.output_path(),123            dest_file_name124        )125        trydelete(dest)126        with open(dest, 'w') as fp:127            json.dump(data, fp)128        return dest129    def clean(self) -> None:130        for file in [131            self.encoded_data_file_path(),132            self.model_file_path(),133            self.score_file_path()134        ]:135            trydelete(file)136        for directory in [137            self.metadata_path(),138            self.output_path(),139        ]:140            clean_dir(directory)141        logging.info('Cleaned output and metadata')142    def run_all(self, run_id=None, clean=True) -> Dict:143        if run_id:144            self.run_id = run_id145        result = {146            'algorithm_name': self.algorithm_definition['algorithm_name'],147            'run_id': self.run_id,148            'ran_at': self.ran_at,149            'algorithm_definition': self.algorithm_definition150        }151        self.clean()152        result['states'] = self.generate_states()153        result['package_encode_params'] = self.package_encode_parameters()154        result['encode'] = self.encode()155        result['train'] = self.train()156        result['package_predict_params'] = self.package_predict_parameters()157        result['predict'] = self.predict()158        result['evaluate'] = self.evaluate()159        if clean:160            self.clean_output()161        return result162    def _base_command(self, metadata_location: str) -> List[str]:163        ret = [164            'java',165            '-cp', f"{self.hotvect_util_jar_path}",166        ]167        ret.extend(self.jvm_options)168        ret.extend(['com.eshioji.hotvect.commandline.Main',169                    '--algorithm-jar', f'{self.algorithm_jar_path}',170                    '--algorithm-definition', self._write_algorithm_definition(),171                    '--meta-data', metadata_location])172        return ret173    def generate_states(self) -> Dict:174        states = self.algorithm_definition.get('vectorizer_parameters', {}).get('feature_states', {})175        metadata = {}176        for state_name, instruction in states.items():177            metadata_path = os.path.join(self.metadata_path(), f'generate-state-{state_name}.json')178            trydelete(metadata_path)179            output_path = self.state_output_path(state_name)180            trydelete(output_path)181            source_path = os.path.join(self.state_source_base_path, instruction['source_name'])182            ensure_file_exists(source_path)183            generation_task = instruction['generation_task']184            if instruction.get('cache'):185                cached = os.path.join(self.state_source_base_path, instruction['cache'])186                ensure_file_exists(cached)187                logging.info(f'Using cache for state:{state_name} from {cached}')188                copyfile(cached, output_path)189                metadata[state_name] = {190                    'cache': cached191                }192            else:193                logging.info(f'No cache found for state:{state_name}, generating')194                cmd = self._base_command(metadata_path)195                cmd.extend(['--generate-state', generation_task,196                            '--training-data', self.train_data_path,197                            '--source', source_path,198                            '--dest', output_path])199                runshell(cmd)200                metadata[state_name] = read_json(metadata_path)201            self.feature_states[state_name] = output_path202        return metadata203    def package_encode_parameters(self) -> Dict:204        encode_parameter_package_location = self.encode_parameter_file_path()205        trydelete(encode_parameter_package_location)206        to_package = list(self.feature_states.values())207        to_zip_archive(to_package, encode_parameter_package_location)208        return {209            'sources': to_package,210            'package': encode_parameter_package_location211        }212    def encode(self) -> Dict:213        metadata_location = os.path.join(self.metadata_path(), 'encode_metadata.json')214        trydelete(metadata_location)215        encoded_data_location = self.encoded_data_file_path()216        trydelete(encoded_data_location)217        cmd = self._base_command(metadata_location)218        cmd.append('--encode')219        if self.feature_states:220            # We have feature states221            cmd.extend(['--parameters', self.encode_parameter_file_path()])222        cmd.extend(['--source', self.train_data_path])223        cmd.extend(['--dest', encoded_data_location])224        runshell(cmd)225        return read_json(metadata_location)226    def train(self) -> Dict:227        metadata_location = os.path.join(self.metadata_path(), 'train_metadata.json')228        trydelete(metadata_location)229        model_location = self.model_file_path()230        trydelete(model_location)231        cmd = [232            'vw', self.encoded_data_file_path(),233            '--readable_model', model_location,234            '--noconstant',235            '--loss_function', 'logistic',236        ]237        train_params = self.algorithm_definition['training_parameters']238        cmd.extend(train_params)239        train_log = runshell(cmd)240        metadata = {241            'training_parameters': train_params,242            'train_log': train_log243        }244        with open(metadata_location, 'w') as f:245            json.dump(metadata, f)246        return metadata247    def package_predict_parameters(self) -> Dict:248        predict_parameter_package_location = self.predict_parameter_file_path()249        trydelete(predict_parameter_package_location)250        # Add the model file251        to_package = [self.model_file_path()]252        # Add all the feature states253        to_package.extend(list(self.feature_states.values()))254        # Add the algo parameters255        algorithm_parameters = {256            'algorithm_name': self.algorithm_definition['algorithm_name'],257            'parameter_id': self.run_id,258            'ran_at': self.ran_at,259            'algorithm_definition': self.algorithm_definition,260            'sources': to_package,261            'package': predict_parameter_package_location262        }263        algo_parameters_path = self._write_data(algorithm_parameters, 'algorithm_parameters.json')264        to_package.append(algo_parameters_path)265        to_zip_archive(to_package, predict_parameter_package_location)266        return algorithm_parameters267    def predict(self) -> Dict:268        metadata_location = os.path.join(self.metadata_path(), 'predict_metadata.json')269        trydelete(metadata_location)270        score_location = self.score_file_path()271        trydelete(score_location)272        cmd = self._base_command(metadata_location)273        cmd.append('--predict')274        cmd.extend(['--source', self.validation_data_location])275        cmd.extend(['--dest', score_location])276        cmd.extend(['--parameters', self.predict_parameter_file_path()])277        runshell(cmd)278        return read_json(metadata_location)279    def evaluate(self) -> Dict:280        metadata_location = os.path.join(self.metadata_path(), 'evaluate_metadata.json')281        trydelete(metadata_location)282        df = pd.read_csv(self.score_file_path(), header=None)283        lower_auc, mean_auc, upper_auc = mlutils.bootstrap_roc_auc(df[1], df[0])284        meta_data = {285            'upper_auc': upper_auc,286            'mean_auc': mean_auc,287            'lower_auc': lower_auc288        }289        with open(metadata_location, 'w') as f:290            json.dump(meta_data, f)291        return meta_data292    def clean_output(self) -> None:293        trydelete(self.encoded_data_file_path())294        trydelete(self.model_file_path())295        trydelete(self.score_file_path())296        trydelete(self.output_path())297    def audit(self) -> None:298        metadata_location = os.path.join(self.metadata_path(), 'audit_metadata.json')299        trydelete(metadata_location)300        audit_data_location = self.audit_data_file_path()301        trydelete(audit_data_location)302        cmd = self._base_command(metadata_location)303        cmd.append('--audit')304        if self.feature_states:305            # We have feature states306            cmd.extend(['--parameters', self.encode_parameter_file_path()])307        cmd.extend(['--source', self.train_data_path])308        cmd.extend(['--dest', audit_data_location])309        runshell(cmd)...

influxdb_client.py

Source:influxdb_client.py

1"""InfluxDBClient is client for API defined in https://github.com/influxdata/influxdb/blob/master/http/swagger.yml."""2from __future__ import absolute_import3import configparser4import os5from influxdb_client import Configuration, ApiClient, HealthCheck, HealthService, Ready, ReadyService6from influxdb_client.client.authorizations_api import AuthorizationsApi7from influxdb_client.client.bucket_api import BucketsApi8from influxdb_client.client.delete_api import DeleteApi9from influxdb_client.client.labels_api import LabelsApi10from influxdb_client.client.organizations_api import OrganizationsApi11from influxdb_client.client.query_api import QueryApi12from influxdb_client.client.tasks_api import TasksApi13from influxdb_client.client.users_api import UsersApi14from influxdb_client.client.write_api import WriteApi, WriteOptions, PointSettings15class InfluxDBClient(object):16    """InfluxDBClient is client for InfluxDB v2."""17    def __init__(self, url, token, debug=None, timeout=10_000, enable_gzip=False, org: str = None,18                 default_tags: dict = None, **kwargs) -> None:19        """20        Initialize defaults.21        :param url: InfluxDB server API url (ex. http://localhost:8086).22        :param token: auth token23        :param debug: enable verbose logging of http requests24        :param timeout: HTTP client timeout setting for a request specified in milliseconds.25                        If one number provided, it will be total request timeout.26                        It can also be a pair (tuple) of (connection, read) timeouts.27        :param enable_gzip: Enable Gzip compression for http requests. Currently only the "Write" and "Query" endpoints28                            supports the Gzip compression.29        :param org: organization name (used as a default in query and write API)30        :key bool verify_ssl: Set this to false to skip verifying SSL certificate when calling API from https server.31        :key str ssl_ca_cert: Set this to customize the certificate file to verify the peer.32        :key str proxy: Set this to configure the http proxy to be used (ex. http://localhost:3128)33        :key int connection_pool_maxsize: Number of connections to save that can be reused by urllib3.34                                          Defaults to "multiprocessing.cpu_count() * 5".35        :key urllib3.util.retry.Retry retries: Set the default retry strategy that is used for all HTTP requests36                                               except batching writes. As a default there is no one retry strategy.37        """38        self.url = url39        self.token = token40        self.org = org41        self.default_tags = default_tags42        conf = _Configuration()43        if self.url.endswith("/"):44            conf.host = self.url[:-1]45        else:46            conf.host = self.url47        conf.enable_gzip = enable_gzip48        conf.debug = debug49        conf.verify_ssl = kwargs.get('verify_ssl', True)50        conf.ssl_ca_cert = kwargs.get('ssl_ca_cert', None)51        conf.proxy = kwargs.get('proxy', None)52        conf.connection_pool_maxsize = kwargs.get('connection_pool_maxsize', conf.connection_pool_maxsize)53        conf.timeout = timeout54        auth_token = self.token55        auth_header_name = "Authorization"56        auth_header_value = "Token " + auth_token57        retries = kwargs.get('retries', False)58        self.api_client = ApiClient(configuration=conf, header_name=auth_header_name,59                                    header_value=auth_header_value, retries=retries)60    def __enter__(self):61        """62        Enter the runtime context related to this object.63        It will bind this methodâs return value to the target(s)64        specified in the `as` clause of the statement.65        return: self instance66        """67        return self68    def __exit__(self, exc_type, exc_value, traceback):69        """Exit the runtime context related to this object and close the client."""70        self.close()71    @classmethod72    def from_config_file(cls, config_file: str = "config.ini", debug=None, enable_gzip=False):73        """74        Configure client via configuration file. The configuration has to be under 'influx' section.75        The supported formats:76            - https://docs.python.org/3/library/configparser.html77            - https://toml.io/en/78        Configuration options:79            - url80            - org81            - token82            - timeout,83            - verify_ssl84            - ssl_ca_cert85            - connection_pool_maxsize86        config.ini example::87            [influx2]88            url=http://localhost:808689            org=my-org90            token=my-token91            timeout=600092            connection_pool_maxsize=2593            [tags]94            id = 132-987-65595            customer = California Miner96            data_center = ${env.data_center}97        config.toml example::98            [influx2]99                url = "http://localhost:8086"100                token = "my-token"101                org = "my-org"102                timeout = 6000103                connection_pool_maxsize = 25104            [tags]105                id = "132-987-655"106                customer = "California Miner"107                data_center = "${env.data_center}"108        """109        config = configparser.ConfigParser()110        config.read(config_file)111        def config_value(key: str):112            return config['influx2'][key].strip('"')113        url = config_value('url')114        token = config_value('token')115        timeout = None116        if config.has_option('influx2', 'timeout'):117            timeout = config_value('timeout')118        org = None119        if config.has_option('influx2', 'org'):120            org = config_value('org')121        verify_ssl = True122        if config.has_option('influx2', 'verify_ssl'):123            verify_ssl = config_value('verify_ssl')124        ssl_ca_cert = None125        if config.has_option('influx2', 'ssl_ca_cert'):126            ssl_ca_cert = config_value('ssl_ca_cert')127        connection_pool_maxsize = None128        if config.has_option('influx2', 'connection_pool_maxsize'):129            connection_pool_maxsize = config_value('connection_pool_maxsize')130        default_tags = None131        if config.has_section('tags'):132            tags = {k: v.strip('"') for k, v in config.items('tags')}133            default_tags = dict(tags)134        return cls(url, token, debug=debug, timeout=_to_int(timeout), org=org, default_tags=default_tags,135                   enable_gzip=enable_gzip, verify_ssl=_to_bool(verify_ssl), ssl_ca_cert=ssl_ca_cert,136                   connection_pool_maxsize=_to_int(connection_pool_maxsize))137    @classmethod138    def from_env_properties(cls, debug=None, enable_gzip=False):139        """140        Configure client via environment properties.141        Supported environment properties:142            - INFLUXDB_V2_URL143            - INFLUXDB_V2_ORG144            - INFLUXDB_V2_TOKEN145            - INFLUXDB_V2_TIMEOUT146            - INFLUXDB_V2_VERIFY_SSL147            - INFLUXDB_V2_SSL_CA_CERT148            - INFLUXDB_V2_CONNECTION_POOL_MAXSIZE149        """150        url = os.getenv('INFLUXDB_V2_URL', "http://localhost:8086")151        token = os.getenv('INFLUXDB_V2_TOKEN', "my-token")152        timeout = os.getenv('INFLUXDB_V2_TIMEOUT', "10000")153        org = os.getenv('INFLUXDB_V2_ORG', "my-org")154        verify_ssl = os.getenv('INFLUXDB_V2_VERIFY_SSL', "True")155        ssl_ca_cert = os.getenv('INFLUXDB_V2_SSL_CA_CERT', None)156        connection_pool_maxsize = os.getenv('INFLUXDB_V2_CONNECTION_POOL_MAXSIZE', None)157        default_tags = dict()158        for key, value in os.environ.items():159            if key.startswith("INFLUXDB_V2_TAG_"):160                default_tags[key[16:].lower()] = value161        return cls(url, token, debug=debug, timeout=_to_int(timeout), org=org, default_tags=default_tags,162                   enable_gzip=enable_gzip, verify_ssl=_to_bool(verify_ssl), ssl_ca_cert=ssl_ca_cert,163                   connection_pool_maxsize=_to_int(connection_pool_maxsize))164    def write_api(self, write_options=WriteOptions(), point_settings=PointSettings()) -> WriteApi:165        """166        Create a Write API instance.167        :param point_settings:168        :param write_options: write api configuration169        :return: write api instance170        """171        return WriteApi(influxdb_client=self, write_options=write_options, point_settings=point_settings)172    def query_api(self) -> QueryApi:173        """174        Create a Query API instance.175        :return: Query api instance176        """177        return QueryApi(self)178    def close(self):179        """Shutdown the client."""180        self.__del__()181    def __del__(self):182        """Shutdown the client."""183        if self.api_client:184            self.api_client.__del__()185            self.api_client = None186    def buckets_api(self) -> BucketsApi:187        """188        Create the Bucket API instance.189        :return: buckets api190        """191        return BucketsApi(self)192    def authorizations_api(self) -> AuthorizationsApi:193        """194        Create the Authorizations API instance.195        :return: authorizations api196        """197        return AuthorizationsApi(self)198    def users_api(self) -> UsersApi:199        """200        Create the Users API instance.201        :return: users api202        """203        return UsersApi(self)204    def organizations_api(self) -> OrganizationsApi:205        """206        Create the Organizations API instance.207        :return: organizations api208        """209        return OrganizationsApi(self)210    def tasks_api(self) -> TasksApi:211        """212        Create the Tasks API instance.213        :return: tasks api214        """215        return TasksApi(self)216    def labels_api(self) -> LabelsApi:217        """218        Create the Labels API instance.219        :return: labels api220        """221        return LabelsApi(self)222    def health(self) -> HealthCheck:223        """224        Get the health of an instance.225        :return: HealthCheck226        """227        health_service = HealthService(self.api_client)228        try:229            health = health_service.get_health()230            return health231        except Exception as e:232            return HealthCheck(name="influxdb", message=str(e), status="fail")233    def ready(self) -> Ready:234        """235        Get The readiness of the InfluxDB 2.0.236        :return: Ready237        """238        ready_service = ReadyService(self.api_client)239        return ready_service.get_ready()240    def delete_api(self) -> DeleteApi:241        """242        Get the delete metrics API instance.243        :return: delete api244        """245        return DeleteApi(self)246class _Configuration(Configuration):247    def __init__(self):248        Configuration.__init__(self)249        self.enable_gzip = False250    def update_request_header_params(self, path: str, params: dict):251        super().update_request_header_params(path, params)252        if self.enable_gzip:253            # GZIP Request254            if path == '/api/v2/write':255                params["Content-Encoding"] = "gzip"256                params["Accept-Encoding"] = "identity"257                pass258            # GZIP Response259            if path == '/api/v2/query':260                # params["Content-Encoding"] = "gzip"261                params["Accept-Encoding"] = "gzip"262                pass263            pass264        pass265    def update_request_body(self, path: str, body):266        _body = super().update_request_body(path, body)267        if self.enable_gzip:268            # GZIP Request269            if path == '/api/v2/write':270                import gzip271                if isinstance(_body, bytes):272                    return gzip.compress(data=_body)273                else:274                    return gzip.compress(bytes(_body, "utf-8"))275        return _body276def _to_bool(bool_value):277    return str(bool_value).lower() in ("yes", "true")278def _to_int(int_value):...

configure_caddy.py

Source:configure_caddy.py

1#!/usr/bin/python2"""3Configure caddy service4"""5import os6import sys7import json8import bcrypt9import logging10import coloredlogs11import argparse12from urllib.parse import quote, urljoin13from subprocess   import run, call14import functions as func15### Enable logging16logging.basicConfig(17    format='%(asctime)s [%(levelname)s] %(message)s', 18    level=logging.INFO, 19    stream=sys.stdout)20log = logging.getLogger(__name__)21### Enable argument parsing22parser = argparse.ArgumentParser()23parser.add_argument('--opts', type=json.loads, help='Set script arguments')24parser.add_argument('--env', type=json.loads, help='Set script environment')25parser.add_argument('--user', type=json.loads, help='Load user settings')26parser.add_argument('--settings', type=json.loads, help='Load script settings')27args, unknown = parser.parse_known_args()28if unknown:29    log.error("Unknown arguments " + str(unknown))30### Load arguments31cli_opts = args.opts32cli_env = args.env33cli_user = args.user34cli_settings = args.settings35### Set log level36verbosity = cli_opts.get("verbosity")37log.setLevel(verbosity)38# Setup colored console logs39coloredlogs.install(fmt='%(asctime)s [%(levelname)s] %(message)s', level=verbosity, logger=log)40### Get envs41proxy_base_url = cli_env.get("PROXY_BASE_URL")42caddy_virtual_port = cli_env.get("CADDY_VIRTUAL_PORT")43caddy_virtual_host = cli_env.get("CADDY_VIRTUAL_HOST")44caddy_virtual_proto = cli_env.get("CADDY_VIRTUAL_PROTO")45caddy_virtual_base_url = cli_env.get("CADDY_VIRTUAL_BASE_URL")46caddy_proxy_encodings_gzip = cli_env.get("CADDY_PROXY_ENCODINGS_GZIP")47caddy_proxy_encodings_zstd = cli_env.get("CADDY_PROXY_ENCODINGS_ZSTD")48caddy_proxy_templates = cli_env.get("CADDY_PROXY_TEMPLATES")49caddy_letsencrypt_email = cli_env.get("CADDY_LETSENCRYPT_EMAIL")50caddy_letsencrypt_endpoint = cli_env.get("CADDY_LETSENCRYPT_ENDPOINT")51caddy_http_port = cli_env.get("CADDY_HTTP_PORT")52caddy_https_port = cli_env.get("CADDY_HTTPS_PORT")53caddy_auto_https = cli_env.get("CADDY_AUTO_HTTPS")54fb_port = cli_user.get("filebrowser").get("port")55fb_base_url = cli_user.get("filebrowser").get("base_url")56vscode_bind_addr = cli_user.get("vscode").get("bind_addr")57vscode_base_url = cli_user.get("vscode").get("base_url")58app_bind_addr = cli_user.get("app").get("bind_addr")59app_base_url = cli_user.get("app").get("base_url")60### Get user settings61user_name = cli_user.get("name")62user_group = cli_user.get("group")63user_home = cli_user.get("dirs").get("home").get("path")64### Clean up envs65application = "caddy"66proxy_base_url = func.clean_url(proxy_base_url)67host_fqdn = caddy_virtual_host # @TODO: Not reading from env68host_port = caddy_virtual_port69host_ip = "0.0.0.0"70host_proto = caddy_virtual_proto71host_base_url = func.clean_url(caddy_virtual_base_url)72auto_https = True if caddy_auto_https == "true" else False73enable_gzip = True if caddy_proxy_encodings_gzip == "true" else False74enable_zstd = True if caddy_proxy_encodings_zstd == "true" else False75enable_templates = True if caddy_proxy_templates == "true" else False76### Set config and data paths77config_dir = os.path.join(user_home, ".config", application)78if not os.path.exists(config_dir):79    os.makedirs(config_dir)80storage = os.path.join(config_dir, "storage")81if not os.path.exists(storage): 82    os.mkdir(storage)83### Set certificate endpoint84letsencrypt_staging = "https://acme-staging-v02.api.letsencrypt.org/directory"85letsencrypt_production = "https://acme-v02.api.letsencrypt.org/directory"86if caddy_letsencrypt_endpoint == "dev":87    endpoint = letsencrypt_staging88elif caddy_letsencrypt_endpoint == "prod":89    endpoint = letsencrypt_production90elif caddy_letsencrypt_endpoint == "internal":91    #@TODO: Get internal certs working92    endpoint = letsencrypt_production = "set this up"93else:94    log.info(f"invalid letsencrypt endpoint: '{caddy_letsencrypt_endpoint}'")95### Run protocol check96if not host_proto in ['http', 'https']:97    log.critical(f"{application}: protocol '{proto}' is not valid! Exiting.")98    sys.exit()99### Define application route settings100servers = dict()101servers["automatic_https"]: auto_https102servers['default'] = dict()103domains = dict()104domains[host_fqdn] = ""105vscode_settings = {106    "name": "vscode",107    "host": "localhost",108    "port": vscode_bind_addr.split(":",1)[1],109    "proto": "http",110    "base_url": func.clean_url(vscode_base_url),111    "enable_gzip": True,112    "enable_gzip": True,113    "enable_templates": True,114}115filebrowser_settings = {116    "name": "filebrowser",117    "host": "localhost",118    "port": fb_port,119    "proto": "http",120    "base_url": func.clean_url(fb_base_url),121    "enable_gzip": True,122    "enable_gzip": True,123    "enable_templates": True,124}125app_settings = {126    "name": "app",127    "host": "localhost",128    "port": app_bind_addr.split(":",1)[1],129    "proto": "http",130    "base_url": func.clean_url(app_base_url),131    "enable_gzip": True,132    "enable_gzip": True,133    "enable_templates": True,134}135### Create application sub-config templates136service_settings = [vscode_settings, filebrowser_settings, app_settings]137subroutes = list()138for service in service_settings:139    service_base_url = urljoin(host_base_url, service.get("base_url"))140    full_base_url = urljoin(proxy_base_url, service_base_url) if service_base_url != "/" else ""141    log.info("{name} base url: '{url}'".format(name=service.get("name"), url=full_base_url))142    encodings = dict()143    if service.get("enable_gzip") or service.get("enable_zstd"):144        encodings = {145            "handle": [{146                "encodings": {},147                "handler": "encode"148            }]149        }150        if service.get("enable_gzip"):151            encodings["handle"][0]["encodings"]['gzip'] = dict()152        if service.get("enable_zstd"):153            encodings["handle"][0]["encodings"]['zstd']= dict()154    templates = dict()155    if service.get("enable_templates"):156        templates = {157            "handle": [{158                "handler": "templates"159            }]160        }161    subroute = {162                "handler": "subroute",163                "routes": [{164                    "handle": [{165                            "handler": "static_response",166                            "headers": {167                                "Location": [168                                f"{full_base_url}/"169                                ]170                            },171                            "status_code": 302172                        }173                    ],174                    "match": [{175                            "path": [176                                f"{full_base_url}"177                            ]178                        }179                    ]180                },181                {182                    "handle": [{183                            "handler": "subroute",184                            "routes": [{185                                "handle": [{186                                    "handler": "rewrite",187                                    "strip_path_prefix": f"{full_base_url}"188                                }]189                            },190                            {191                            "handle": [{192                                "handler": "reverse_proxy",193                                "upstreams": [{194                                "dial": "{}:{}".format(service.get("host"), service.get("port"))195                                }]196                            }]197                            },198                            encodings,199                            templates200                            ]201                        }],202                    "match": [{203                        "path": [204                            f"{full_base_url}/*"205                        ]206                    }]207                }]208            }209    subroutes.append(subroute)210if host_fqdn != None:211    if host_fqdn == "":212        match = []213    else:214        match = [{          215            "host": [host_fqdn]216        }]217    route = {218        "match": match,219        "handle": subroutes,220        "terminal": True221    }222if servers['default'].get('routes') == None:223    servers['default']['listen'] = [f"{host_ip}:{host_port}"]224    servers['default']['routes'] = [route]225    servers['default']['logs'] = {226        "logger_names": {227            host_fqdn: "common",228        }229    }230else:231    servers['default']['routes'].append(route)232### Create config template233config_file = {234    "admin": {235        "disabled": False,236        "listen": '',237        "enforce_origin": False,238        "origins": [''],239        "config": {240            "persist": False241        }242    },243    "logging": {244		"logs": {245            "default": {246                "exclude": [247                    "http.log.access.json",248                    "http.log.access.common",249                    "http.log.access.common_and_json"250                ]251            },252			"common": {253				"writer": {254                    "output": "stdout"255                },256                "encoder": {257                    "format": "single_field",258                    "field": "common_log"259                },260				"level": "",261				"sampling": {262					"interval": 0,263					"first": 0,264					"thereafter": 0265				},266				"include": ["http.log.access.common"],267			}268		}269    },270    "storage": {271        "module": "file_system",272	    "root": storage273    },274    "apps": {275        "http": {276            "http_port": int(caddy_http_port),277            "https_port": int(caddy_https_port),278            "servers": servers279        },280        "tls": {281            "automation": {282                "policies": [{283                    "subjects": list(domains.keys()),284                    "issuers": [285                        {286                        "module": "acme",287                        "ca": endpoint,288                        "email": caddy_letsencrypt_email289                        },290                        {291                        "module": "internal",292                        "ca": "",293                        "lifetime": 0,294                        "sign_with_root": False295                        }296                    ],297                    "key_type": ""298                }]299            }300        }301    }302}303### Write config file304config_path = os.path.join(config_dir, "settings.json")305config_json = json.dumps(config_file, indent = 4)306with open(config_path, "w") as f: 307    f.write(config_json)308# fix permissions309log.info(f"setting permissions on '{config_dir}' to '{user_name}:{user_group}'")310func.recursive_chown(config_dir, user_name, user_group)311### Display final config312log.debug(f"{application} config: '{config_path}'")...

Playwright tutorial

LambdaTest’s Playwright tutorial will give you a broader idea about the Playwright automation framework, its unique features, and use cases with examples to exceed your understanding of Playwright testing. This tutorial will give A to Z guidance, from installing the Playwright framework to some best practices and advanced concepts.

Chapters:

What is Playwright : Playwright is comparatively new but has gained good popularity. Get to know some history of the Playwright with some interesting facts connected with it.
How To Install Playwright : Learn in detail about what basic configuration and dependencies are required for installing Playwright and run a test. Get a step-by-step direction for installing the Playwright automation framework.
Playwright Futuristic Features: Launched in 2020, Playwright gained huge popularity quickly because of some obliging features such as Playwright Test Generator and Inspector, Playwright Reporter, Playwright auto-waiting mechanism and etc. Read up on those features to master Playwright testing.
What is Component Testing: Component testing in Playwright is a unique feature that allows a tester to test a single component of a web application without integrating them with other elements. Learn how to perform Component testing on the Playwright automation framework.
Inputs And Buttons In Playwright: Every website has Input boxes and buttons; learn about testing inputs and buttons with different scenarios and examples.
Functions and Selectors in Playwright: Learn how to launch the Chromium browser with Playwright. Also, gain a better understanding of some important functions like “BrowserContext,” which allows you to run multiple browser sessions, and “newPage” which interacts with a page.
Handling Alerts and Dropdowns in Playwright : Playwright interact with different types of alerts and pop-ups, such as simple, confirmation, and prompt, and different types of dropdowns, such as single selector and multi-selector get your hands-on with handling alerts and dropdown in Playright testing.
Playwright vs Puppeteer: Get to know about the difference between two testing frameworks and how they are different than one another, which browsers they support, and what features they provide.
Run Playwright Tests on LambdaTest: Playwright testing with LambdaTest leverages test performance to the utmost. You can run multiple Playwright tests in Parallel with the LammbdaTest test cloud. Get a step-by-step guide to run your Playwright test on the LambdaTest platform.
Playwright Python Tutorial: Playwright automation framework support all major languages such as Python, JavaScript, TypeScript, .NET and etc. However, there are various advantages to Python end-to-end testing with Playwright because of its versatile utility. Get the hang of Playwright python testing with this chapter.
Playwright End To End Testing Tutorial: Get your hands on with Playwright end-to-end testing and learn to use some exciting features such as TraceViewer, Debugging, Networking, Component testing, Visual testing, and many more.
Playwright Video Tutorial: Watch the video tutorials on Playwright testing from experts and get a consecutive in-depth explanation of Playwright automation testing.