wonderbeyond

#!/usr/bin/env python
import argparse
import urllib.parse
import textwrap

import psycopg2
import psycopg2.extras


def main():

wonderbeyond

from typing import Any

import sqlalchemy as sa
from sqlalchemy.dialects.postgresql import array as pg_array, JSONB


def jsonb_set(target, path: list[str], val: Any):
    """
    An easy wrapper over sa.func.jsonb_set().

wonderbeyond

import os
from contextlib import contextmanager
from tempfile import mkstemp


@contextmanager
def MakeTemporaryFile(suffix=None, prefix=None, dir=None):
    """An alternative to tempfile.NamedTemporaryFile working with Windows NT."""
    try:
        fd, name = mkstemp(suffix=suffix, prefix=prefix, dir=dir)

wonderbeyond

"""
A Task-Step model that can describe steps' upstream and downstream relationships.
Also refer to https://docs.sqlalchemy.org/en/20/orm/join_conditions.html#self-referential-many-to-many-relationship
"""
from typing import Annotated, List

import sqlalchemy as sa
from sqlalchemy import ForeignKey
from sqlalchemy.orm import Mapped, mapped_column, relationship, backref
from sqlalchemy.orm import MappedAsDataclass, DeclarativeBase

wonderbeyond

class AsyncSQLAlchemy:
    def __init__(
        self,
        url: str,
        engine_options: Optional[Dict[str, Any]] = None,
        session_options: Optional[Dict[str, Any]] = None,
    ):
        from sqlalchemy.orm import sessionmaker
        from sqlalchemy.ext.asyncio import AsyncSession

wonderbeyond

{
    "env": {
        "browser": true,
        "es2021": true,
        "commonjs": true,
        "jest/globals": true
    },
    "plugins": ["jest"],
    "extends": "eslint:recommended",
    "overrides": [

wonderbeyond

/**
* Performs a deep merge of objects and returns new object. Does not modify
* objects (immutable) and merges arrays via concatenation.
*
* @param {...object} objects - Objects to merge
* @returns {object} New object with merged key/values
*/
export default function mergeDeep(...objects) {
  const isObject = obj => obj && typeof obj === 'object';

wonderbeyond

/**
 *
 * @param {Object[]} dataset
 * @param {string} key
 * @returns {Object.<string, Object[]>}
 * @example groupBy([
 *   {scope: 'step:1', name: 'API'},
 *   {scope: 'step:1', name: 'amount'},
 *   {scope: 'step:2', name: 'Solvent'}
 * ], 'scope') => {

wonderbeyond

Make a Custom JSON Encoder/Serializer

A example of custom json serializer:

# file: json.py
def custom_json_encoder(o):
    if isinstance(o, pydantic.BaseModel):
        res = o.dict()
 return res['__root__'] if '__root__' in res else res

wonderbeyond

from typing import cast
import logging
import json
import urllib.request
import http.client
import urllib.parse

logger = logging.getLogger(__name__)