normanlmfung

# https://docs.python.org/3/library/contextlib.html
	
from contextlib import contextmanager 
class MessageWriter(object):
  def __init__(self, filename):
    self.file_name = filename

  @contextmanager
  def open_file(self):
    try:

normanlmfung

# Taken DIRECTLY from William's article https://realpython.com/python-interface/
import abc

class FormalParserInterface(metaclass=abc.ABCMeta):
  @classmethod
  def __subclasshook__(cls, subclass):
    print("FormalParserInterface.__subclasshook__")
    return (hasattr(subclass, 'load_data_source') and 
        callable(subclass.load_data_source) and 
        hasattr(subclass, 'extract_text') and 

normanlmfung

# timer (Simply log the duration of time the wrapped method took). Can also be extended to filter/modify parameters.
def timer(method):
	def timed(*args, **kw):
		start = time.time()
		result = method(*args, **kw)
		elapsed = (time.time() - start) * 1000 

		logger.info(f'@timer {method.__qualname__} returned in {elapsed} ms. {args_}')

		return result

normanlmfung

# REST API guard: check api key (If incorrect API key, raise exception)
def check_api_key(api_key : str, request : Any, symmetric_key : str, kms_gizmo : Any):
			def decorator(method):
				headers = request.headers
				encrypted_api_key_from_header = headers.get("X-Api-Key")
				decrypted_api_key_from_header = encrypted_api_key_from_header
				if kms_gizmo:
					decrypted_api_key_from_header = kms_gizmo.decrypt(decrypted_api_key_from_header)
				if symmetric_key:
					decrypted_api_key_from_header = cipher_gizmo.decrypt(encrypted_text=decrypted_api_key_from_header, key=symmetric_key)

normanlmfung

from collections import defaultdict

word_lists = defaultdict(list)

pairs = [('a', 1), ('b', 2), ('c', 3), ('a', 4), ('b', 5)]

for key, value in pairs:
  word_lists[key].append(value) # Append value to list

print(word_lists.get('d', []))

normanlmfung

# This is taken DIRECTLY from Geekforgeeks https://www.geeksforgeeks.org/heap-queue-or-heapq-in-python/
	import heapq

	li = [5, 7, 9, 1, 3]
	# If you call heapify, when you call heappop, order 1,3,5,7,8. If you comment out heapify call, heappop oder 5,7,9,1,3
	heapq.heapify(li) 
	while li:
		print(heapq.heappop(li))

	# Example 2 with tuples: (w, n) where w = weight, n = node id. Heapify uses 'w' to sort first, then 'n' as tie-breaker if there are two/more elements with same 'w'.

normanlmfung

import concurrent.futures
import time

def background_task(task_id):
  item_list = []
  for i in range(3):
    item_list.append((task_id, i))
    time.sleep(1)
  return item_list

normanlmfung

import concurrent.futures
import time

def background_task1():
  item_list = []
  for i in range(10):
    item_list.append(i)
    time.sleep(1)

  # Simulate an exception being thrown

normanlmfung

'''
https://superfastpython.com/asyncio-semaphore/
https://superfastpython.com/thread-semaphore/
https://superfastpython.com/multiprocessing-semaphore-in-python/

There're three versions of Semaphore:
  from asyncio import Semaphore
  from threading import Semaphore
  from multiprocessing import Semaphore

normanlmfung

nums = [ 1,2,3,4,5 ]
squares = []
for num in nums:
  squares.append(num**2)
# instead of loop, use 'map'
squares = list(map(lambda num: num ** 2, nums))