statuspage/app/ultrametrics.py

import asyncio, logging, time, aiohttp
from datetime import datetime
from threading import Thread
from typing import Optional, Tuple, Sequence
from app import crud
from app.database import SessionLocal


logging.basicConfig(format='%(levelname)s:%(message)s', filename='example.log', filemode='w', level=logging.INFO)

db=SessionLocal()

URLS = [
	"https://python.org",
	"https://google.com",
	"https://stackoverflow.com",
	'https://dotya.ml',
	'https://git.dotya.ml',
	'https://drone.dotya.ml',
	'https://netdata.dotya.ml',
	'https://tew0x00.dotya.ml',
	'https://speedtest.dotya.ml',
	'https://testdotya.dotya.ml',
	"https://ubuntu.com",
]
URL = crud.get_urls(db=db)

def start_background_loop(loop: asyncio.AbstractEventLoop) -> None:
	asyncio.set_event_loop(loop)
	loop.run_forever()


async def fetch(url: str, session: aiohttp.ClientSession = None) -> Tuple[str, str]:
	async def _fetch(url: str, session: aiohttp.ClientSession):
		URL = crud.get_urls(db=db)
		s = time.time()
		async with session.head(url) as response:
			f = time.time() - s
			m = ('[ %f ]\t<%s>\ttook: %f\t(%s)' %(time.time(),response.status,f,url))
			print(m)
			logging.info(m)
			return url


	if session:
		return await _fetch(url, session)
	else:
		async with aiohttp.ClientSession() as session:
			return await _fetch(url, session)

async def fetch_urls(loop: asyncio.AbstractEventLoop) -> Sequence[Tuple[str, str]]:
	async with aiohttp.ClientSession() as session:
		tasks = [loop.create_task(fetch(url, session)) for url in URL]
		results = await asyncio.gather(*tasks)
		return results

def main() -> None:
	loop = asyncio.new_event_loop()
	t = Thread(target=start_background_loop, args=(loop,), daemon=True)
	t.start()



	while True:
		task = asyncio.run_coroutine_threadsafe(fetch_urls(loop), loop)
		time.sleep(10)


if __name__ == '__main__':
	main()