Mirrors/chatmail-core
1
0
Fork 0
mirror of https://github.com/chatmail/core.git synced 2026-04-06 07:32:12 +03:00
Code Issues Packages Projects Releases Wiki Activity

update python API (#3394)

Browse Source
This commit is contained in:
Asiel Díaz Benítez
2022-06-04 12:12:38 -04:00
committed by GitHub
parent 0d1d1a25da
commit 4f02c811a3
6 changed files with 304 additions and 54 deletions
Download Patch File Download Diff File Expand all files Collapse all files

59
python/src/deltachat/account.py
View File

@@ -62,12 +62,15 @@ class Account(object):
MissingCredentials = MissingCredentials
def __init__(self, db_path, os_name=None, logging=True) -> None:
def __init__(self, db_path, os_name=None, logging=True, closed=False) -> None:
"""initialize account object.
:param db_path: a path to the account database. The database
will be created if it doesn't exist.
:param os_name: this will be put to the X-Mailer header in outgoing messages
:param os_name: [Deprecated]
:param logging: enable logging for this account
:param closed: set to True to avoid automatically opening the account
after creation.
"""
# initialize per-account plugin system
self._pm = hookspec.PerAccount._make_plugin_manager()
@@ -80,7 +83,7 @@ class Account(object):
db_path = db_path.encode("utf8")
self._dc_context = ffi.gc(
lib.dc_context_new(as_dc_charpointer(os_name), db_path, ffi.NULL),
lib.dc_context_new_closed(db_path) if closed else lib.dc_context_new(ffi.NULL, db_path, ffi.NULL),
lib.dc_context_unref,
)
if self._dc_context == ffi.NULL:
@@ -92,6 +95,17 @@ class Account(object):
hook = hookspec.Global._get_plugin_manager().hook
hook.dc_account_init(account=self)
def open(self, passphrase: Optional[str] = None) -> bool:
"""Open the account's database with the given passphrase.
This can only be used on a closed account. If the account is new, this
operation sets the database passphrase. For existing databases the passphrase
should be the one used to encrypt the database the first time.
:returns: True if the database is opened with this passphrase, False if the
passphrase is incorrect or an error occurred.
"""
return bool(lib.dc_context_open(self._dc_context, as_dc_charpointer(passphrase)))
def disable_logging(self) -> None:
"""disable logging."""
self._logging = False
@@ -209,13 +223,13 @@ class Account(object):
:returns: True if account is configured.
"""
return True if lib.dc_is_configured(self._dc_context) else False
return bool(lib.dc_is_configured(self._dc_context))
def is_open(self) -> bool:
"""Determine if account is open
:returns True if account is open."""
return True if lib.dc_context_is_open(self._dc_context) else False
return bool(lib.dc_context_is_open(self._dc_context))
def set_avatar(self, img_path: Optional[str]) -> None:
"""Set self avatar.
@@ -461,21 +475,24 @@ class Account(object):
"""
return self._export(path, imex_cmd=const.DC_IMEX_EXPORT_SELF_KEYS)
def export_all(self, path):
"""return new file containing a backup of all database state
(chats, contacts, keys, media, ...). The file is created in the
the `path` directory.
def export_all(self, path: str, passphrase: Optional[str] = None) -> str:
"""Export a backup of all database state (chats, contacts, keys, media, ...).
:param path: the directory where the backup will be stored.
:param passphrase: the backup will be encrypted with the passphrase, if it is
None or empty string, the backup is not encrypted.
:returns: path to the created backup.
Note that the account has to be stopped; call stop_io() if necessary.
"""
export_files = self._export(path, const.DC_IMEX_EXPORT_BACKUP)
export_files = self._export(path, const.DC_IMEX_EXPORT_BACKUP, passphrase)
if len(export_files) != 1:
raise RuntimeError("found more than one new file")
return export_files[0]
def _export(self, path, imex_cmd):
def _export(self, path: str, imex_cmd: int, passphrase: Optional[str] = None) -> list:
with self.temp_plugin(ImexTracker()) as imex_tracker:
self.imex(path, imex_cmd)
self.imex(path, imex_cmd, passphrase)
return imex_tracker.wait_finish()
def import_self_keys(self, path):
@@ -487,21 +504,23 @@ class Account(object):
"""
self._import(path, imex_cmd=const.DC_IMEX_IMPORT_SELF_KEYS)
def import_all(self, path):
"""import delta chat state from the specified backup `path` (a file).
def import_all(self, path: str, passphrase: Optional[str] = None) -> None:
"""Import Delta Chat state from the specified backup file.
The account must be in unconfigured state for import to attempted.
:param path: path to the backup file.
:param passphrase: if not None or empty, the backup will be opened with the passphrase.
"""
assert not self.is_configured(), "cannot import into configured account"
self._import(path, imex_cmd=const.DC_IMEX_IMPORT_BACKUP)
self._import(path, imex_cmd=const.DC_IMEX_IMPORT_BACKUP, passphrase=passphrase)
def _import(self, path, imex_cmd):
def _import(self, path: str, imex_cmd: int, passphrase: Optional[str] = None) -> None:
with self.temp_plugin(ImexTracker()) as imex_tracker:
self.imex(path, imex_cmd)
self.imex(path, imex_cmd, passphrase)
imex_tracker.wait_finish()
def imex(self, path, imex_cmd):
lib.dc_imex(self._dc_context, imex_cmd, as_dc_charpointer(path), ffi.NULL)
def imex(self, path: str, imex_cmd: int, passphrase: Optional[str] = None) -> None:
lib.dc_imex(self._dc_context, imex_cmd, as_dc_charpointer(path), as_dc_charpointer(passphrase))
def initiate_key_transfer(self) -> str:
"""return setup code after a Autocrypt setup message

102
python/src/deltachat/chat.py
View File

@@ -65,27 +65,65 @@ class Chat(object):
# ------ chat status/metadata API ------------------------------
def is_group(self) -> bool:
"""return true if this chat is a group chat.
"""Return True if this chat is a group chat.
:returns: True if chat is a group-chat, false otherwise
:returns: True if chat is a group-chat, False otherwise
"""
return lib.dc_chat_get_type(self._dc_chat) == const.DC_CHAT_TYPE_GROUP
def is_single(self) -> bool:
"""Return True if this chat is a single/direct chat, False otherwise"""
return lib.dc_chat_get_type(self._dc_chat) == const.DC_CHAT_TYPE_SINGLE
def is_mailinglist(self) -> bool:
"""Return True if this chat is a mailing list, False otherwise"""
return lib.dc_chat_get_type(self._dc_chat) == const.DC_CHAT_TYPE_MAILINGLIST
def is_broadcast(self) -> bool:
"""Return True if this chat is a broadcast list, False otherwise"""
return lib.dc_chat_get_type(self._dc_chat) == const.DC_CHAT_TYPE_BROADCAST
def is_multiuser(self) -> bool:
"""Return True if this chat is a multi-user chat (group, mailing list or broadcast), False otherwise"""
return lib.dc_chat_get_type(self._dc_chat) in (
const.DC_CHAT_TYPE_GROUP,
const.DC_CHAT_TYPE_MAILINGLIST,
const.DC_CHAT_TYPE_BROADCAST,
)
def is_self_talk(self) -> bool:
"""Return True if this chat is the self-chat (a.k.a. "Saved Messages"), False otherwise"""
return bool(lib.dc_chat_is_self_talk(self._dc_chat))
def is_device_talk(self) -> bool:
"""Returns True if this chat is the "Device Messages" chat, False otherwise"""
return bool(lib.dc_chat_is_device_talk(self._dc_chat))
def is_muted(self) -> bool:
"""return true if this chat is muted.
:returns: True if chat is muted, False otherwise.
"""
return lib.dc_chat_is_muted(self._dc_chat)
return bool(lib.dc_chat_is_muted(self._dc_chat))
def is_contact_request(self):
def is_pinned(self) -> bool:
"""Return True if this chat is pinned, False otherwise"""
return lib.dc_chat_get_visibility(self._dc_chat) == const.DC_CHAT_VISIBILITY_PINNED
def is_archived(self) -> bool:
"""Return True if this chat is archived, False otherwise.
:returns: True if archived, False otherwise
"""
return lib.dc_chat_get_visibility(self._dc_chat) == const.DC_CHAT_VISIBILITY_ARCHIVED
def is_contact_request(self) -> bool:
"""return True if this chat is a contact request chat.
:returns: True if chat is a contact request chat, False otherwise.
"""
return lib.dc_chat_is_contact_request(self._dc_chat)
return bool(lib.dc_chat_is_contact_request(self._dc_chat))
def is_promoted(self):
def is_promoted(self) -> bool:
"""return True if this chat is promoted, i.e.
the member contacts are aware of their membership,
have been sent messages.
@@ -100,14 +138,14 @@ class Chat(object):
:returns: True if the chat is writable, False otherwise
"""
return lib.dc_chat_can_send(self._dc_chat)
return bool(lib.dc_chat_can_send(self._dc_chat))
def is_protected(self) -> bool:
"""return True if this chat is a protected chat.
:returns: True if chat is protected, False otherwise.
"""
return lib.dc_chat_is_protected(self._dc_chat)
return bool(lib.dc_chat_is_protected(self._dc_chat))
def get_name(self) -> Optional[str]:
"""return name of this chat.
@@ -125,6 +163,18 @@ class Chat(object):
name = as_dc_charpointer(name)
return bool(lib.dc_set_chat_name(self.account._dc_context, self.id, name))
def get_color(self):
"""return the color of the chat.
:returns: color as 0x00rrggbb
"""
return lib.dc_chat_get_color(self._dc_chat)
def get_summary(self):
"""return dictionary with summary information."""
dc_res = lib.dc_chat_get_info_json(self.account._dc_context, self.id)
s = from_dc_charpointer(dc_res)
return json.loads(s)
def mute(self, duration: Optional[int] = None) -> None:
"""mutes the chat
@@ -148,6 +198,24 @@ class Chat(object):
if not bool(ret):
raise ValueError("Failed to unmute chat")
def pin(self) -> None:
"""Pin the chat."""
lib.dc_set_chat_visibility(self.account._dc_context, self.id, const.DC_CHAT_VISIBILITY_PINNED)
def unpin(self) -> None:
"""Unpin the chat."""
if self.is_pinned():
lib.dc_set_chat_visibility(self.account._dc_context, self.id, const.DC_CHAT_VISIBILITY_NORMAL)
def archive(self) -> None:
"""Archive the chat."""
lib.dc_set_chat_visibility(self.account._dc_context, self.id, const.DC_CHAT_VISIBILITY_ARCHIVED)
def unarchive(self) -> None:
"""Unarchive the chat."""
if self.is_archived():
lib.dc_set_chat_visibility(self.account._dc_context, self.id, const.DC_CHAT_VISIBILITY_NORMAL)
def get_mute_duration(self) -> int:
"""Returns the number of seconds until the mute of this chat is lifted.
@@ -364,12 +432,6 @@ class Chat(object):
"""
return lib.dc_marknoticed_chat(self.account._dc_context, self.id)
def get_summary(self):
"""return dictionary with summary information."""
dc_res = lib.dc_chat_get_info_json(self.account._dc_context, self.id)
s = from_dc_charpointer(dc_res)
return json.loads(s)
# ------ group management API ------------------------------
def add_contact(self, obj):
@@ -460,12 +522,6 @@ class Chat(object):
return None
return from_dc_charpointer(dc_res)
def get_color(self):
"""return the color of the chat.
:returns: color as 0x00rrggbb
"""
return lib.dc_chat_get_color(self._dc_chat)
# ------ location streaming API ------------------------------
def is_sending_locations(self):
@@ -474,12 +530,6 @@ class Chat(object):
"""
return lib.dc_is_sending_locations_to_chat(self.account._dc_context, self.id)
def is_archived(self):
"""return True if this chat is archived.
:returns: True if archived.
"""
return lib.dc_chat_get_visibility(self._dc_chat) == const.DC_CHAT_VISIBILITY_ARCHIVED
def enable_sending_locations(self, seconds):
"""enable sending locations for this chat.

4
python/src/deltachat/message.py
View File

@@ -159,6 +159,10 @@ class Message(object):
"""
return from_dc_charpointer(lib.dc_get_msg_info(self.account._dc_context, self.id))
def get_summarytext(self, width: int) -> str:
"""Get a message summary as a single line of text. Typically used for notifications."""
return from_dc_charpointer(lib.dc_msg_get_summarytext(self._dc_msg, width))
def continue_key_transfer(self, setup_code):
"""extract key and use it as primary key for this account."""
res = lib.dc_continue_key_transfer(self.account._dc_context, self.id, as_dc_charpointer(setup_code))

16
python/src/deltachat/testplugin.py
View File

@@ -11,7 +11,7 @@ import threading
import time
import weakref
from queue import Queue
from typing import Callable, List
from typing import Callable, List, Optional
import pytest
import requests
@@ -429,16 +429,16 @@ class ACFactory:
if addr in self.testprocess._addr2files:
return self._getaccount(addr)
def get_unconfigured_account(self):
return self._getaccount()
def get_unconfigured_account(self, closed=False):
return self._getaccount(closed=closed)
def _getaccount(self, try_cache_addr=None):
def _getaccount(self, try_cache_addr=None, closed=False):
logid = "ac{}".format(len(self._accounts) + 1)
# we need to use fixed database basename for maybe_cache_* functions to work
path = self.tmpdir.mkdir(logid).join("dc.db")
if try_cache_addr:
self.testprocess.cache_maybe_retrieve_configured_db_files(try_cache_addr, path)
ac = Account(path.strpath, logging=self._logging)
ac = Account(path.strpath, logging=self._logging, closed=closed)
ac._logid = logid # later instantiated FFIEventLogger needs this
ac._evtracker = ac.add_account_plugin(FFIEventTracker(ac))
if self.pytestconfig.getoption("--debug-setup"):
@@ -467,9 +467,11 @@ class ACFactory:
else:
print("WARN: could not use preconfigured keys for {!r}".format(addr))
def get_pseudo_configured_account(self):
def get_pseudo_configured_account(self, passphrase: Optional[str] = None) -> Account:
# do a pseudo-configured account
ac = self.get_unconfigured_account()
ac = self.get_unconfigured_account(closed=bool(passphrase))
if passphrase:
ac.open(passphrase)
acname = ac._logid
addr = "{}@offline.org".format(acname)
ac.update_config(