40 lines
1.1 KiB
Python
40 lines
1.1 KiB
Python
|
import environ
|
||
|
|
from celery import Celery
|
||
|
|
from pywireguard.factory import Peer
|
||
|
|
from .wg_manager import WGManager
|
||
|
|
|
||
|
|
|
||
|
|
env = environ.Env(WG_INTERFACE=(str, "wg0"), CELERY_BROKER=(str, "redis://localhost"), CELERY_BACKEND=(str, "redis://localhost"))
|
||
|
|
environ.Env.read_env(".env")
|
||
|
|
WG_INTERFACE = env("WG_INTERFACE")
|
||
|
|
CELERY_BROKER = env("CELERY_BROKER")
|
||
|
|
CELERY_BACKEND = env("CELERY_BACKEND")
|
||
|
|
app = Celery("wg_manager_tasks", broker=CELERY_BROKER, backend=CELERY_BACKEND)
|
||
|
|
app.conf.event_serializer = "pickle" # this event_serializer is optional. somehow i missed this when writing this solution and it still worked without.
|
||
|
|
app.conf.task_serializer = "pickle"
|
||
|
|
app.conf.result_serializer = "pickle"
|
||
|
|
app.conf.accept_content = ["application/json", "application/x-python-serialize"]
|
||
|
|
|
||
|
|
|
||
|
|
@app.task(ignore_result=True)
|
||
|
|
def add_peer(peer: Peer):
|
||
|
|
wg = WGManager(WG_INTERFACE)
|
||
|
|
wg.add_peer(peer)
|
||
|
|
|
||
|
|
|
||
|
|
@app.task(ignore_result=True)
|
||
|
|
def remove_peer(peer: Peer):
|
||
|
|
wg = WGManager(WG_INTERFACE)
|
||
|
|
wg.remove_peer(peer)
|
||
|
|
|
||
|
|
|
||
|
|
@app.task
|
||
|
|
def get_peers() -> list[Peer]:
|
||
|
|
wg = WGManager(WG_INTERFACE)
|
||
|
|
return wg.get_peers()
|
||
|
|
|
||
|
|
|
||
|
|
@app.task
|
||
|
|
def ping() -> str:
|
||
|
|
return "pong"
|