Small aiohttp server application for TLE storage
Project description
TLEStorageService
Small aiohttp server application for TLE storage
Requirements
aio-nasa-tle-loader >= 1.0.0
aio-space-track-api >= 2.1.1
aiohttp >= 2.1.0
aiopg >= 0.13.0
alembic >= 0.9.2
pyaml >=16.12.2
sqlalchemy >= 1.1.10
ujson >= 1.35
Installing
pip install tle-storage-service
Getting started
Start aiohttp application:
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import logging
import os
from aiohttp import web
from tle_storage_service import TleStorageService, load_cfg, LOG_FORMAT
DIR = os.path.abspath(os.path.dirname(__file__))
if __name__ == '__main__':
logging.basicConfig(level=logging.INFO, format=LOG_FORMAT)
app = TleStorageService(config=load_cfg(path=os.path.join(DIR, 'config', 'server.yaml')))
web.run_app(app, port=8080)
Create periodic script for refresh db
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import asyncio
import logging
import os
from aio_nasa_tle_loader import AsyncNasaTLELoader
from aio_space_track_api import AsyncSpaceTrackApi
from aiopg.sa import create_engine
from tle_storage_service import periodic_task, load_cfg, nasa, space_track, LOG_FORMAT
DIR = os.path.abspath(os.path.dirname(__file__))
CFG = load_cfg(path=os.path.join(DIR, 'config', 'periodic.yaml'))
logger = logging.getLogger(__name__)
@periodic_task(CFG['sync'].get('space-track', 4) * 3600)
async def call_space_track(engine, api_params, satellites, loop):
async with AsyncSpaceTrackApi(loop=loop, **api_params) as api:
await space_track(engine, api, satellites)
@periodic_task(CFG['sync'].get('nasa', 0.5) * 3600)
async def call_nasa(engine, loop):
async with AsyncNasaTLELoader(loop=loop) as loader:
await nasa(engine, loader)
async def run(db_params, api_params, satellites, loop):
async with create_engine(loop=loop, **db_params) as engine:
await asyncio.gather(
call_nasa(engine, loop),
call_space_track(engine, api_params, satellites, loop),
loop=loop
)
def notify_callback(msg):
logger.info('[%s] Receive msg <- "%s"', msg.channel, msg.payload)
if __name__ == '__main__':
logging.basicConfig(format=LOG_FORMAT, level=logging.DEBUG)
loop = asyncio.get_event_loop()
loop.create_task(run(CFG['db'],
CFG['space-track'],
CFG['sync']['satellites'],
loop))
try:
loop.run_forever()
finally:
loop.close()
More code examples into examples directory
Source code
The latest developer version is available in a github repository: https://github.com/nkoshell/tle-storage-service
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
tle-storage-service-1.0.3.tar.gz
(14.8 kB
view hashes)
Built Distribution
Close
Hashes for tle-storage-service-1.0.3.tar.gz
Algorithm | Hash digest | |
---|---|---|
SHA256 | 49341b104d55365d421919dad1f43a4eba81f4c05162887962fa6213565e2b88 |
|
MD5 | 7bace1005b69df5d6b1483d8df3c4479 |
|
BLAKE2b-256 | 1afab8dd4b9afa35d4538c58766440e0bdedeb6b08f6184b71960db79b27f376 |
Close
Hashes for tle_storage_service-1.0.3-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | 7a0081c9aa19f07128e4dc695cbaab0b01dd5040f38454d316ed289d824d68db |
|
MD5 | d0a0b01d4d4b8304866d45c625ef7b19 |
|
BLAKE2b-256 | d34b7fbdc1b240575a9ffce913ff1e0027c4e5a66108f2c26c26fefac638bd99 |