ArthurDelannoyazerty

import time
import ntplib
import logging

logger = logging.getLogger(__name__)

def get_current_utc_time() -> float:
    logger.debug('Requesting current UTC time')
    response = ntplib.NTPClient().request('pool.ntp.org')
    logger.debug(f'UTC time : {response.tx_time}  |  {time.strftime('%Y-%m-%dT%H-%M-%S',time.gmtime(response.tx_time))}')

ArthurDelannoyazerty

import argparse

parser = argparse.ArgumentParser()
parser.add_argument('minute_threshold', type=int)
args = parser.parse_args()
minute_threshold:int = args.minute_threshold

ArthurDelannoyazerty

List to table

We transpose the list so that the Nth column is the Nth list of the entire array Column n = l[n-1])

headers = ['Column 1', 'Column 2', 'Column 3']
l = [[1, 2, 3],
     [4, 5, 6],
     [7, 8, 9]]

table = tabulate.tabulate(list(map(list, zip(*l))), headers=headers, tablefmt="github")

ArthurDelannoyazerty

import asyncio
import logging
import os
import psycopg2
import threading
import time

from dotenv import find_dotenv, load_dotenv

def listen_to_notifications():

ArthurDelannoyazerty

import asyncio
import psycopg2

# dbname should be the same for the notifying process
conn = psycopg2.connect(host="localhost", dbname="example", user="example", password="example")
conn.set_isolation_level(psycopg2.extensions.ISOLATION_LEVEL_AUTOCOMMIT)

cursor = conn.cursor()
cursor.execute(f"LISTEN match_updates;")

ArthurDelannoyazerty

import os

from dotenv import find_dotenv, load_dotenv
from flask import Flask, request, jsonify, render_template
from flask_cors import CORS
from flask_socketio import SocketIO, emit, send

# LOAD ENV VARIABLES -----------------------------------------------------------------------------------
load_dotenv(find_dotenv())
backend_ip = os.environ.get("BACKEND_IP")

ArthurDelannoyazerty

import os
import logging
import boto3

from tabulate import tabulate
from dataclasses import dataclass
from dotenv import find_dotenv, load_dotenv

logger = logging.getLogger(__name__)

ArthurDelannoyazerty

import psycopg2

conn = psycopg2.connect(database="database",
                        host="host",
                        port="port",
                        user="user",
                        password="password")


def _send_query(query_no_variable:str, variable:tuple|None=None, results:bool=False) -> list[tuple]:

ArthurDelannoyazerty

import logging
from datetime import datetime

def setup_logging(log_name:str, 
                  log_dir:str='logs/',
                  level_library = logging.INFO,
                  level_project = logging.DEBUG):
    """Setup the logging global parameters.

    :param str log_name: The name of the gile that contains the logs for this session

ArthurDelannoyazerty

import logging
import os

from llama_index.embeddings.ollama import OllamaEmbedding
from llama_index.llms.ollama import Ollama
from llama_index.core import Settings
from enum import StrEnum
from dotenv import find_dotenv, load_dotenv