Mark Mishyn mark-mishyn

web monkey

mark-mishyn / sqs_long_polling.py

Created February 29, 2024 07:49

SQS messages processing using long polling, it a very lightweight alternative for Celery

	import json
	import threading

	import boto3

	from configurations.config_store import Config
	from src.routes.frontapp.tasks import (
	add_tags_to_conversation,
	reply_to_front_message,
	)

mark-mishyn / generate_function_json_schema.py

Last active June 1, 2024 16:30

Simple example of generation JSON Schema for OpenAI function_call with Python and Pydandic

	import json

	import openai
	from pydantic import BaseModel, PrivateAttr

	from src.common.config import Config

	openai.api_key = Config.OPEN_AI_API_KEY

mark-mishyn / sqs_send_batch.py

Created September 20, 2021 08:23

Send SQS messages in batch

	class BatchSendToSQS:
	"""
	>>> with BatchSendToSQS(someQueue, batch_size=10) as sqs_batcher:
	>>> sqs_batcher.add_msg({'data': 'booba'})
	>>> ...
	>>> sqs_batcher.add_msg({'data': 'piu piu'})
	"""

	def __init__(self, queue: "boto3.sqs.Queue", batch_size: int):
	self.queue = queue

mark-mishyn / get_responses_async.py

Last active September 7, 2021 15:22

Make multiple requests in async and return responses in key-value (key-response) format

	import asyncio

	import aiohttp
	from multidict import CIMultiDictProxy

	@dataclass
	class AsyncToSyncResponse:
	status: int
	content: dict
	headers: CIMultiDictProxy

mark-mishyn / alchemy_print_SQL.py

Last active June 13, 2021 15:19

Just a "SQL printer" for SQLAlchemy. It's a colored example for a SQLA ORM Query.

	import sqlparse
	from pygments import highlight
	from pygments.formatters.terminal import TerminalFormatter
	from pygments.lexers import SqlLexer # to highlight SQL statements
	from sqlalchemy import create_engine
	from sqlalchemy.orm import Query

	engine = create_engine("sqlite+pysqlite:///db.sqlite", echo=True, future=True)

	def format_sql(query: Query):

mark-mishyn / tabler.py

Last active January 28, 2021 09:15

Usefull to pretty print Django models or querysets

	def tabl(model_or_qs, field_names: str = '', limit: int = 1000):
	"""
	Examples:
	>>> tabl(UserStatus)
	or
	>>> tabl(User, 'first_name,last_name')
	or
	>>> tabl(User.objects.filter(is_staff=True), 'first_name,last_name', limit=20)
	"""
	qs = model_or_qs.objects.all() if inspect.isclass(model_or_qs) else model_or_qs

mark-mishyn / signleton_model.py

Last active March 18, 2020 08:07

Django Singleton Model and ModelAdmin

	from django.db import models
	from django.contrib import admin
	from django.contrib.admin.options import csrf_protect_m


	class SingletonModel(models.Model):

	class Meta:
	abstract = True

mark-mishyn / extract_body_from_gmail_mail.py

Last active December 5, 2019 07:32

Extracts body of gmail mail

	'''
	Example of usage:
	from googleapiclient.discovery import build
	from httplib2 import Http
	from oauth2client import file

	# Create google client first, for example
	storage = file.Storage('path_to_google_oauth_credentials')
	creds = storage.get()
	http = creds.authorize(Http())

mark-mishyn / seconds_to_iso8601_duration.py

Last active March 4, 2020 15:58

	from datetime import datetime

	def seconds_to_iso8601_duration(seconds: int) -> str:
	return datetime.utcfromtimestamp(seconds).strftime('PT%HH%MM%SS')

mark-mishyn / flatten.py

Last active March 8, 2019 18:35

Convert dict to flat dict

	import collections

	def flatten(d: dict, parent_key='', sep='__', flat_lists=False) -> dict:
	"""
	>>> flatten({'name': 'Ivan', 'mother': {'name': 'Alisha'}})
	{'name': 'Ivan', 'mother__name': 'Alisha'}

	>>> flatten({'name': 'Ivan', 'brothers': [{'name': 'Jack'}]}, flat_lists=True)
	{'name': 'Ivan', 'brothers_0__name': 'Jack'}