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
def make_chunks(text, chunk_size: int, overlap: int = 0): | |
"""Split text into chunks with overlap.""" | |
chunks = [] | |
len_ = len(text) | |
for i in range(0, len_, chunk_size - overlap): | |
chunks.append(text[i: i + chunk_size]) | |
if i + chunk_size >= len_: | |
break | |
return chunks |
#!/usr/bin/env python | |
import argparse | |
import urllib.parse | |
import textwrap | |
import psycopg2 | |
import psycopg2.extras | |
def main(): |
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(). |
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) |
""" | |
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 |
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 |
{ | |
"env": { | |
"browser": true, | |
"es2021": true, | |
"commonjs": true, | |
"jest/globals": true | |
}, | |
"plugins": ["jest"], | |
"extends": "eslint:recommended", | |
"overrides": [ |
/** | |
* 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'; |
/** | |
* | |
* @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') => { |