How to use the atomicwrites.atomic_write function in atomicwrites

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

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

github spyder-ide / spyder / spyder / utils / encoding.py View on Github external
text, encoding = encode(text, encoding)
    if 'a' in mode:
        with open(filename, mode) as textfile:
            textfile.write(text)
    else:
        # Based in the solution at untitaker/python-atomicwrites#42.
        # Needed to fix file permissions overwritting.
        # See spyder-ide/spyder#9381.
        try:
            original_mode = os.stat(filename).st_mode
        except OSError:  # Change to FileNotFoundError for PY3
            # Creating a new file, emulate what os.open() does
            umask = os.umask(0)
            os.umask(umask)
            original_mode = 0o777 & ~umask
        with atomic_write(filename,
                          overwrite=True,
                          mode=mode) as textfile:
            textfile.write(text)
        os.chmod(filename, original_mode)
    return encoding
github pimutils / vdirsyncer / vdirsyncer / storage / google.py View on Github external
def _save_token(token):
            checkdir(expand_path(os.path.dirname(token_file)), create=True)
            with atomic_write(token_file, mode='w', overwrite=True) as f:
                json.dump(token, f)
github gridsync / gridsync / gridsync / tahoe.py View on Github external
upload_dircap_ro = upload_dircap_json[1]["ro_uri"]
        yield self.link(admin_dircap, "admin", upload_dircap_ro)
        yaml_path = os.path.join(self.nodedir, "private", "magic_folders.yaml")
        try:
            with open(yaml_path) as f:
                yaml_data = yaml.safe_load(f)
        except OSError:
            yaml_data = {}
        folders_data = yaml_data.get("magic-folders", {})
        folders_data[os.path.basename(path)] = {
            "directory": path,
            "collective_dircap": collective_dircap,
            "upload_dircap": upload_dircap,
            "poll_interval": poll_interval,
        }
        with atomic_write(yaml_path, mode="w", overwrite=True) as f:
            f.write(yaml.safe_dump({"magic-folders": folders_data}))
        self.add_alias(alias, admin_dircap)
github jbms / finance-dl / finance_dl / ultipro_google.py View on Github external
lambda: self.find_element_in_any_frame(
                        By.XPATH,
                        '//input[@type="image" and contains(@title, "download")]'
                    ))
                download_link.click()
                logger.info('%s: Waiting to get download', document_str)
                download_result, = self.wait_and_return(
                    self.get_downloaded_file)
                name, data = download_result
                if len(data) < 5000:
                    raise RuntimeError(
                        'Downloaded file size is invalid: %d' % len(data))
                output_name = '%s.statement-%s.pdf' % (
                    pay_date.strftime('%Y-%m-%d'), document_number)
                output_path = os.path.join(self.output_directory, output_name)
                with atomic_write(output_path, mode='wb') as f:
                    f.write(data)
                downloaded_statements.add((pay_date, document_number))
                return True
            else:
                logger.info('%s: Just downloaded', document_str)
        return False
github jbms / finance-dl / finance_dl / google_purchases.py View on Github external
def _fetch_html_pages(self, need_to_fetch: List[Tuple[str, str]]):
        logger.info('Fetching details for %d purchases', len(need_to_fetch))
        for i, (purchase_id, html_path) in enumerate(need_to_fetch):
            url = 'https://myaccount.google.com/purchases/detail?order_id=' + purchase_id
            logger.info('Fetching details %d/%d: %s', i, len(need_to_fetch), url)
            with self.wait_for_page_load():
                self.driver.get(url)
            content = self.driver.page_source
            with atomic_write(
                    html_path, mode='w', encoding='utf-8', newline='\n') as f:
                # Write with Unicode Byte Order Mark to ensure content will be properly interpreted as UTF-8
                f.write('\ufeff' + content)
            logger.info('Write details %d/%d: %s', i, len(need_to_fetch), html_path)
github apriha / snps / src / snps / resources.py View on Github external
def _download_assembly_mapping_data(
        self, destination, chroms, source_assembly, target_assembly, retries
    ):
        with atomic_write(destination, mode="wb", overwrite=True) as f:
            with tarfile.open(fileobj=f, mode="w:gz") as out_tar:
                for chrom in chroms:
                    file = chrom + ".json"

                    map_endpoint = "/map/human/{}/{}/{}?".format(
                        source_assembly, chrom, target_assembly
                    )

                    # get assembly mapping data
                    response = None
                    retry = 0
                    while response is None and retry < retries:
                        response = self._ensembl_rest_client.perform_rest_action(
                            map_endpoint
                        )
                        retry += 1
github jbms / finance-dl / finance_dl / paypal.py View on Github external
newline='\n') as f:
                        f.write(json.dumps(transaction, indent='  '))
                continue
            details_url = (
                'https://www.paypal.com/myaccount/transactions/details/' +
                transaction_id)
            inline_details_url = (
                'https://www.paypal.com/myaccount/transactions/details/inline/'
                + transaction_id)
            html_path = output_prefix + '.html'
            json_path = output_prefix + '.json'
            if not os.path.exists(html_path):
                logging.info('Retrieving HTML %s', details_url)
                html_resp = self.driver.request('GET', details_url)
                html_resp.raise_for_status()
                with atomic_write(
                        html_path, mode='w', encoding='utf-8',
                        newline='\n') as f:
                    # Write with Unicode Byte Order Mark to ensure content will be properly interpreted as UTF-8
                    f.write('\ufeff' + html_resp.text)
            if not os.path.exists(json_path):
                logging.info('Retrieving JSON %s', inline_details_url)
                json_resp = self.make_json_request(inline_details_url)
                json_resp.raise_for_status()
                j = json_resp.json()
                jsonschema.validate(j, transaction_details_schema)
                with atomic_write(json_path, mode='wb') as f:
                    f.write(
                        json.dumps(j['data']['details'], indent='  ').encode())

atomicwrites

Atomic file writes.

MIT
Latest version published 2 years ago

Package Health Score

64 / 100
Full package analysis

Similar packages