etherpad refactoring start
This commit is contained in:
@@ -0,0 +1 @@
|
||||
from .api import getPadHTML, setPadHTML, createPadifNotExists
|
||||
|
||||
@@ -5,18 +5,29 @@ import urllib.parse
|
||||
|
||||
from etherpad_lite import EtherpadLiteClient, EtherpadException
|
||||
|
||||
from contextlib import contextmanager
|
||||
|
||||
|
||||
import logging
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
SERVER_URL = settings.ETHERPAD_CLIENT["exturl"]
|
||||
|
||||
@contextmanager
|
||||
def ignore_ep_exception(msg="", *exceptions):
|
||||
try:
|
||||
yield
|
||||
except exceptions as e:
|
||||
logger.error(msg)
|
||||
logger.error("%s", e)
|
||||
|
||||
def get_ep_client():
|
||||
|
||||
def get_ep_client(groupName="fet"):
|
||||
epc = None
|
||||
group = None
|
||||
|
||||
try:
|
||||
with ignore_ep_exception("Error connecting to Etherpad", Exception):
|
||||
with open(os.path.abspath(settings.ETHERPAD_CLIENT["apikey"]), "r") as f:
|
||||
apikey = f.read()
|
||||
apikey = apikey.rstrip()
|
||||
@@ -27,73 +38,52 @@ def get_ep_client():
|
||||
base_url=urllib.parse.urljoin(settings.ETHERPAD_CLIENT["url"], "api"),
|
||||
api_version="1.2.14",
|
||||
)
|
||||
group = epc.createGroupIfNotExistsFor(groupMapper="fet")
|
||||
except Exception as e:
|
||||
logger.info("Can't get connection to Etherpad Server. Error: {}".format(e))
|
||||
raise e
|
||||
return None, None
|
||||
group = epc.createGroupIfNotExistsFor(groupMapper=groupName)
|
||||
|
||||
return epc, group
|
||||
|
||||
@contextmanager
|
||||
def ep_client(groupName="fet"):
|
||||
epc, group = get_ep_client(groupName)
|
||||
if not epc or not group:
|
||||
yield None, None
|
||||
else:
|
||||
with ignore_ep_exception("EtherpadException!!", EtherpadException):
|
||||
yield epc, group
|
||||
|
||||
|
||||
def __checkPadExists(padID=None):
|
||||
if not padID:
|
||||
return False
|
||||
|
||||
epc, group = get_ep_client()
|
||||
if not epc or not group:
|
||||
return None
|
||||
|
||||
try:
|
||||
lists = epc.listAllPads()
|
||||
except Exception as e:
|
||||
raise e
|
||||
else:
|
||||
string = group["groupID"] + "$" + str(padID)
|
||||
if string in lists["padIDs"]:
|
||||
with ep_client() as (epc, group):
|
||||
if not epc or not group:
|
||||
return None
|
||||
lists = epc.listPads(groupID=group["groupID"])
|
||||
if str(padID) in lists["padIDs"]:
|
||||
return True
|
||||
|
||||
return False
|
||||
|
||||
|
||||
def createPadifNotExists(padID):
|
||||
epc, group = get_ep_client()
|
||||
if not epc:
|
||||
logger.info("if not epc")
|
||||
return None
|
||||
|
||||
# Pad doesn't exist
|
||||
if not __checkPadExists(padID=padID):
|
||||
try:
|
||||
with ep_client() as (epc, group):
|
||||
if not epc or not group:
|
||||
return None
|
||||
epc.createGroupPad(
|
||||
groupID=group["groupID"], padName=padID, text="helloworld"
|
||||
groupID = group["groupID"], padName=padID, text="helloworld"
|
||||
)
|
||||
except EtherpadException as e:
|
||||
logger.info("Can't create Pad '{}'. EtherpadException: {}".format(padID, e))
|
||||
return None
|
||||
except Exception as e:
|
||||
raise e
|
||||
return None
|
||||
|
||||
return padID
|
||||
|
||||
|
||||
def getPadHTML(padID):
|
||||
epc, group = get_ep_client()
|
||||
if not epc or not group:
|
||||
return None
|
||||
|
||||
try:
|
||||
text = epc.getHTML(padID=group["groupID"] + "$" + padID)["html"]
|
||||
except EtherpadException as e:
|
||||
logger.info(
|
||||
"Can't get HTML from padID '{}'. EtherpadException: {}".format(padID, e)
|
||||
)
|
||||
return None
|
||||
except Exception as e:
|
||||
raise e
|
||||
return None
|
||||
|
||||
text = None
|
||||
with ep_client() as (epc, group):
|
||||
if not epc or not group:
|
||||
return None
|
||||
text = epc.getHTML(padID = group["groupID"] + "$" + padID)["html"]
|
||||
return text
|
||||
|
||||
|
||||
@@ -109,11 +99,9 @@ def setPadHTML(padID, html):
|
||||
def get_pad_link(padID):
|
||||
if padID is None:
|
||||
return "#"
|
||||
|
||||
epc, group = get_ep_client()
|
||||
if not epc or not group:
|
||||
return "#"
|
||||
|
||||
return urllib.parse.urljoin(
|
||||
settings.ETHERPAD_CLIENT["exturl"], "p/" + group["groupID"] + "$" + str(padID)
|
||||
)
|
||||
with ep_client() as (epc, group):
|
||||
if not epc or not group:
|
||||
return "#"
|
||||
return urllib.parse.urljoin(
|
||||
settings.ETHERPAD_CLIENT["exturl"], "p/" + group["groupID"] + "$" + str(padID)
|
||||
)
|
||||
|
||||
@@ -10,7 +10,7 @@ from django.utils import timezone
|
||||
from django.utils.text import slugify
|
||||
from django.utils.translation import gettext_lazy as _
|
||||
from taggit.managers import TaggableManager
|
||||
from documents.api import getPadHTML, setPadHTML, createPadifNotExists
|
||||
from documents import getPadHTML, setPadHTML, createPadifNotExists
|
||||
from .managers import (
|
||||
PostManager,
|
||||
ArticleManager,
|
||||
|
||||
23
fet2020/tests/test_etherpad.py
Normal file
23
fet2020/tests/test_etherpad.py
Normal file
@@ -0,0 +1,23 @@
|
||||
import pytest
|
||||
from django.urls import reverse
|
||||
from django.core.validators import ValidationError
|
||||
import datetime
|
||||
from documents.api import ep_client, createPadifNotExists
|
||||
|
||||
|
||||
class TestEtherpad:
|
||||
def test(self):
|
||||
with ep_client() as (epc, group):
|
||||
if not epc:
|
||||
return
|
||||
pds = epc.listPads(groupID=str(group["groupID"]))
|
||||
assert isinstance(pds, dict)
|
||||
|
||||
def test_listpads(self):
|
||||
pds = None
|
||||
with ep_client() as (epc, group):
|
||||
pds = epc.listPads(groupID=group["groupID"])
|
||||
assert isinstance(pds, dict)
|
||||
|
||||
def test_padexists(self):
|
||||
assert createPadifNotExists("sdf")
|
||||
Reference in New Issue
Block a user