Source code for dna.utils

from dna.utils.certbot_utils import Certbot
from dna.utils.db_utils import SQLite, Service, Domain, ApiKey
from dna.utils.docker_utils import Docker
from dna.utils.nginx_utils import Nginx, Block
from dna.utils.log_utils import Logger
from dna.utils.flask_utils import create_api_client, create_logs_client

import subprocess


[docs]def sh(*args, stream=True, **kwargs): """A wrapper around ``subprocess.Popen`` that returns a generator streaming output from the command specified by ``args`` :param args: the command to run, split on whitespaces :param stream: whether to stream the output as a generator (defaults to\ ``True``) :type stream: bool :param kwargs: other options to pass into ``Popen``, such as a ``cwd`` :return: a generator to stream lines from the subprocess output if stream\ is ``True``, else the subprocess output as a completed string """ if not stream: return subprocess.run(args, capture_output=True).stdout.decode("utf-8") out = subprocess.Popen( args, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, universal_newlines=True, bufsize=1, **kwargs, ) def generator(): while True: line = out.stdout.readline() yield line returncode = out.poll() if returncode is not None: return f"Process completed with code {returncode}." return generator()