-
Notifications
You must be signed in to change notification settings - Fork 16
/
yunicode.py
78 lines (60 loc) · 2.02 KB
/
yunicode.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
# Copyright (c) Jupyter Development Team.
# Distributed under the terms of the Modified BSD License.
from functools import partial
from typing import Any, Callable, Optional
from pycrdt import Doc, Text
from .ybasedoc import YBaseDoc
class YUnicode(YBaseDoc):
"""
Extends :class:`YBaseDoc`, and represents a plain text document, encoded as UTF-8.
Schema:
.. code-block:: json
{
"state": YMap,
"source": YText
}
"""
def __init__(self, ydoc: Optional[Doc] = None):
"""
Constructs a YUnicode.
:param ydoc: The :class:`pycrdt.Doc` that will hold the data of the document, if provided.
:type ydoc: :class:`pycrdt.Doc`, optional.
"""
super().__init__(ydoc)
self._ydoc["source"] = self._ysource = Text()
@property
def version(self) -> str:
"""
Returns the version of the document.
:return: Document's version.
:rtype: str
"""
return "1.0.0"
def get(self) -> str:
"""
Returns the content of the document.
:return: Document's content.
:rtype: str
"""
return str(self._ysource)
def set(self, value: str) -> None:
"""
Sets the content of the document.
:param value: The content of the document.
:type value: str
"""
with self._ydoc.transaction():
# clear document
self._ysource.clear()
# initialize document
if value:
self._ysource += value
def observe(self, callback: Callable[[str, Any], None]) -> None:
"""
Subscribes to document changes.
:param callback: Callback that will be called when the document changes.
:type callback: Callable[[str, Any], None]
"""
self.unobserve()
self._subscriptions[self._ystate] = self._ystate.observe(partial(callback, "state"))
self._subscriptions[self._ysource] = self._ysource.observe(partial(callback, "source"))