urllib, tqdm

Progressbar가 있는 파일 다운로드

class DownloadProgressBar(tqdm):
    def update_to(self, b=1, bsize=1, tsize=None):
        if tsize is not None:
            self.total = tsize
        self.update(b * bsize - self.n)
def download_url(url, output_path):
    with DownloadProgressBar(unit='B', unit_scale=True, miniters=1, desc=url.split('/')[-1]) as t:
        urllib.request.urlretrieve(url, filename=output_path, reporthook=t.update_to)

출처 https://stackoverflow.com/a/62113293/10754732

request과 tqdm.contrib.logging 사용하는 버전

logging과 섞어써도 된다는 장점이 있다.

import requests, pathlib, functools, shutil
from tqdm.auto import tqdm
from tqdm.contrib.logging import logging_redirect_tqdm

def download(url, filename, *args, **kwargs):
    r = requests.get(url, stream=True, allow_redirects=True, *args, **kwargs)
    if r.status_code != 200:
        r.raise_for_status()  # Will only raise for 4xx codes, so...
        raise RuntimeError(f"Request to {url} returned status code {r.status_code}")
    file_size = int(r.headers.get('Content-Length', 0))

    path = pathlib.Path(filename).expanduser().resolve()
    path.parent.mkdir(parents=True, exist_ok=True)

    desc = "(Unknown total file size)" if file_size == 0 else ""
    r.raw.read = functools.partial(r.raw.read, decode_content=True)  # Decompress if needed
    with logging_redirect_tqdm():
        with tqdm.wrapattr(r.raw, "read", total=file_size, desc=desc) as r_raw:
            with path.open("wb") as f:
                shutil.copyfileobj(r_raw, f)

    return path

Pyside6

Stdout to GUI redirect

class StdoutWrapper(QThread):
    proc = Signal(str, str)
    daemon = True
    def __init__(self, parent=None):
        super().__init__(parent)
        self.stdout = sys.stdout.write
        self.stderr = sys.stderr.write
    def run(self):
        sys.stdout.write = self.write
        sys.stderr.write = lambda x: self.write(x, color='red')
    def stop(self):
        sys.stdout.write = self.stdout
        sys.stderr.write = self.stderr
        # sys.stdout.write = sys.__stdout__.write
        # sys.stderr.write = sys.__stderr__.write
        self.quit()
        self.wait(1000)
    def write(self, msg, color=None):
        sys.stdout.flush()
        self.proc.emit(msg, color)

메인 윈도우에는 아래와 같은 소스코드를 추가한다.

    def __init__(self, parent=None):
        # 메인 윈도우의 __init__ 부분, 필요 없는 소스코드는 생략
        self.stdout = StdoutWrapper(parent=self)
        self.stdout.proc.connect(self._append_text)
        self.stdout.start()
    def _append_text(self, text, color=None):
        text = text.rstrip()
        if color:
            self.textBrowser.append(f'<span style=" color: {color};">{text}</span>')
        else:
            self.textBrowser.append(text)
    def closeEvent(self, e):
        self.hide()
        self.stdout.terminate()

Worker Thread

class WorkerSignals(QObject):
    finished = Signal()
    result = Signal(object)
class Worker(QRunnable):
    def __init__(self, fn, *args, parent=None, **kwargs):
        super().__init__(parent)
        self.fn = fn
        self.args = args
        self.kwargs = kwargs
        self.signals = WorkerSignals()
    def run(self):
        try:
            result = asyncio.run(self.fn(*self.args, **self.kwargs))
            self.signals.result.emit(result)
            self.signals.finished.emit()
        except:
            print(f"[!] Error while processing {self.fn.__name__}")
            logger.error(f"Error occurred on Worker processing {self.fn}\n{traceback.format_exc()}")

메인 윈도우에 threadpool = QThreadPool.globalInstance()을 정의하고 아래와 같이 사용하면 된다. finished나 result signal은 알아서 연결해서 쓰자.

            worker = Worker(transfer, (인자1), (인자2), (키워드인자)=(키워드인자))
            self.threadpool.start(worker)

SimpleNamespace

dictionary처럼 [] 대괄호를 이용하는 게 아니라, .(온점)을 이용해 .attribute 형식으로 접근 가능한 객체가 필요한데 새 클래스를 정의하기 귀찮다면 SimpleNamespace를 이용하면 된다.

        config = SimpleNamespace(**{})
        config.fromToken = self.fromToken.text()
        config.toToken = self.toToken.text()

아마 **{} 부분은 필요 없을 것 같긴 하다.

Dynamic attribute getter/setter

객체의 attribute를 get/set하고 싶은데 attribute의 이름이 가변적으로 변한다면 Python 내장함수인 getattr, setattr을 사용하면 된다.

class Dummy:
    def __init__(self, name):
        setattr(self, name, 1)

name = "hello" # it will change
d = Dummy(name)
print(getattr(d, name))

Decimal

유효 숫자가 많은 소수(float)를 십진수 체계에서 오차 없이 다루고 싶다면 decimal 라이브러리를 쓰면 된다.

from decimal import Decimal as D
gas = D(1234567890)
gas = gas / D(10) ** D(9) * D("1.2")
print(gas)

Telegram bot with multiprocessing

python-telegram-bot (20.0 이전 버전)을 multiprocessing을 이용해 다루고 싶다면 아래와 같이 구성하면 된다.

class TelegramBot(mp.Process):
    def __new__(cls, *args, **kwargs):
        if not hasattr(cls, 'instance'):
            cls.instance = super().__new__(cls)
        return cls.instance
    def __init__(self, MarketList, SymbolList, isUpdated) -> None:
        mp.Process.__init__(self)
        self.MarketList = MarketList
        self.SymbolList = SymbolList
        self.isUpdated = isUpdated
    def run(self):
        globals()['MarketList'] = self.MarketList
        globals()['SymbolList'] = self.SymbolList
        globals()['isUpdated'] = self.isUpdated
        self.updater = Updater(token = os.environ["TELEGRAM_TOKEN"], use_context = True)
        self.dispatcher = self.updater.dispatcher
        for keyword in ('add', 'delete'):
            self.dispatcher.add_handler(CommandHandler(keyword, getattr(self, keyword)))
            self.dispatcher.add_handler(CommandHandler(keyword[:1], getattr(self, keyword)))
        self.updater.start_polling()
        self.updater.idle()

메인 프로세스에서는 아래 소스코드가 돌아가게 구성하면 된다.

    manager = mp.Manager()
    MarketList = manager.dict()
    SymbolList = manager.dict()
    isUpdated = manager.Value('i', False)
    telegramBot = TelegramBot(MarketList, SymbolList, isUpdated)
    telegramBot.daemon = True
    telegramBot.start()

pathlib

파일 관리할 때 os.path, glob 같은 거 쓰지 말고 Python 3.4부터 생긴 pathlib 쓰자. 훨씬 편리하다. 자체적으로 파일 경로를 자체 클래스로 관리해서 Windows/Linux 상관 없이 동작하게 만들고 Path와 string 사이 / 연산(원래는 나눗셈이나 Path/string은 두 경로를 붙이는 것이 된다.) 등을 지원한다.