mirror of
https://git.selfprivacy.org/SelfPrivacy/selfprivacy-rest-api.git
synced 2024-10-18 03:42:42 +00:00
chore: PR cleanup
This commit is contained in:
parent
c42e2ef3ac
commit
6e0bf4f2a3
|
@ -14,14 +14,12 @@ from selfprivacy_api.migrations.write_token_to_redis import WriteTokenToRedis
|
||||||
from selfprivacy_api.migrations.check_for_system_rebuild_jobs import (
|
from selfprivacy_api.migrations.check_for_system_rebuild_jobs import (
|
||||||
CheckForSystemRebuildJobs,
|
CheckForSystemRebuildJobs,
|
||||||
)
|
)
|
||||||
from selfprivacy_api.migrations.update_services_flake_list import UpdateServicesFlakeList
|
from selfprivacy_api.migrations.add_roundcube import AddRoundcube
|
||||||
from selfprivacy_api.migrations.add_roundcube_to_userdata import AddRoundcubeToUserdata
|
|
||||||
|
|
||||||
migrations = [
|
migrations = [
|
||||||
WriteTokenToRedis(),
|
WriteTokenToRedis(),
|
||||||
CheckForSystemRebuildJobs(),
|
CheckForSystemRebuildJobs(),
|
||||||
UpdateServicesFlakeList(),
|
AddRoundcube(),
|
||||||
AddRoundcubeToUserdata(),
|
|
||||||
]
|
]
|
||||||
|
|
||||||
|
|
||||||
|
|
36
selfprivacy_api/migrations/add_roundcube.py
Normal file
36
selfprivacy_api/migrations/add_roundcube.py
Normal file
|
@ -0,0 +1,36 @@
|
||||||
|
from selfprivacy_api.migrations.migration import Migration
|
||||||
|
|
||||||
|
from selfprivacy_api.services.flake_service_manager import FlakeServiceManager
|
||||||
|
from selfprivacy_api.utils import ReadUserData, WriteUserData
|
||||||
|
|
||||||
|
|
||||||
|
class AddRoundcube(Migration):
|
||||||
|
"""Adds the Roundcube if it is not present."""
|
||||||
|
|
||||||
|
def get_migration_name(self) -> str:
|
||||||
|
return "add_roundcube"
|
||||||
|
|
||||||
|
def get_migration_description(self) -> str:
|
||||||
|
return "Adds the Roundcube if it is not present."
|
||||||
|
|
||||||
|
def is_migration_needed(self) -> bool:
|
||||||
|
with FlakeServiceManager() as manager:
|
||||||
|
if "roundcube" not in manager.services:
|
||||||
|
return True
|
||||||
|
with ReadUserData() as data:
|
||||||
|
if "roundcube" not in data["modules"]:
|
||||||
|
return True
|
||||||
|
return False
|
||||||
|
|
||||||
|
def migrate(self) -> None:
|
||||||
|
with FlakeServiceManager() as manager:
|
||||||
|
if "roundcube" not in manager.services:
|
||||||
|
manager.services[
|
||||||
|
"roundcube"
|
||||||
|
] = "git+https://git.selfprivacy.org/SelfPrivacy/selfprivacy-nixos-config.git?ref=flakes&dir=sp-modules/roundcube"
|
||||||
|
with WriteUserData() as data:
|
||||||
|
if "roundcube" not in data["modules"]:
|
||||||
|
data["modules"]["roundcube"] = {
|
||||||
|
"enable": False,
|
||||||
|
"subdomain": "roundcube",
|
||||||
|
}
|
|
@ -1,24 +0,0 @@
|
||||||
from selfprivacy_api.migrations.migration import Migration
|
|
||||||
|
|
||||||
from selfprivacy_api.utils import ReadUserData, WriteUserData
|
|
||||||
|
|
||||||
|
|
||||||
class AddRoundcubeToUserdata(Migration):
|
|
||||||
"""Add Roundcube to userdata.json if it does not exist"""
|
|
||||||
|
|
||||||
def get_migration_name(self):
|
|
||||||
return "add_roundcube_to_userdata"
|
|
||||||
|
|
||||||
def get_migration_description(self):
|
|
||||||
return "Add Roundcube to userdata.json if it does not exist"
|
|
||||||
|
|
||||||
def is_migration_needed(self):
|
|
||||||
with ReadUserData() as data:
|
|
||||||
if "roundcube" not in data["modules"]:
|
|
||||||
return True
|
|
||||||
|
|
||||||
def migrate(self):
|
|
||||||
with WriteUserData() as data:
|
|
||||||
data["modules"]["roundcube"] = {
|
|
||||||
"subdomain": "roundcube",
|
|
||||||
}
|
|
|
@ -5,13 +5,13 @@ from selfprivacy_api.jobs import JobStatus, Jobs
|
||||||
class CheckForSystemRebuildJobs(Migration):
|
class CheckForSystemRebuildJobs(Migration):
|
||||||
"""Check if there are unfinished system rebuild jobs and finish them"""
|
"""Check if there are unfinished system rebuild jobs and finish them"""
|
||||||
|
|
||||||
def get_migration_name(self):
|
def get_migration_name(self) -> str:
|
||||||
return "check_for_system_rebuild_jobs"
|
return "check_for_system_rebuild_jobs"
|
||||||
|
|
||||||
def get_migration_description(self):
|
def get_migration_description(self) -> str:
|
||||||
return "Check if there are unfinished system rebuild jobs and finish them"
|
return "Check if there are unfinished system rebuild jobs and finish them"
|
||||||
|
|
||||||
def is_migration_needed(self):
|
def is_migration_needed(self) -> bool:
|
||||||
# Check if there are any unfinished system rebuild jobs
|
# Check if there are any unfinished system rebuild jobs
|
||||||
for job in Jobs.get_jobs():
|
for job in Jobs.get_jobs():
|
||||||
if (
|
if (
|
||||||
|
@ -25,8 +25,9 @@ class CheckForSystemRebuildJobs(Migration):
|
||||||
JobStatus.RUNNING,
|
JobStatus.RUNNING,
|
||||||
]:
|
]:
|
||||||
return True
|
return True
|
||||||
|
return False
|
||||||
|
|
||||||
def migrate(self):
|
def migrate(self) -> None:
|
||||||
# As the API is restarted, we assume that the jobs are finished
|
# As the API is restarted, we assume that the jobs are finished
|
||||||
for job in Jobs.get_jobs():
|
for job in Jobs.get_jobs():
|
||||||
if (
|
if (
|
||||||
|
|
|
@ -12,17 +12,17 @@ class Migration(ABC):
|
||||||
"""
|
"""
|
||||||
|
|
||||||
@abstractmethod
|
@abstractmethod
|
||||||
def get_migration_name(self):
|
def get_migration_name(self) -> str:
|
||||||
pass
|
pass
|
||||||
|
|
||||||
@abstractmethod
|
@abstractmethod
|
||||||
def get_migration_description(self):
|
def get_migration_description(self) -> str:
|
||||||
pass
|
pass
|
||||||
|
|
||||||
@abstractmethod
|
@abstractmethod
|
||||||
def is_migration_needed(self):
|
def is_migration_needed(self) -> bool:
|
||||||
pass
|
pass
|
||||||
|
|
||||||
@abstractmethod
|
@abstractmethod
|
||||||
def migrate(self):
|
def migrate(self) -> None:
|
||||||
pass
|
pass
|
||||||
|
|
|
@ -1,26 +0,0 @@
|
||||||
from selfprivacy_api.migrations.migration import Migration
|
|
||||||
from selfprivacy_api.jobs import JobStatus, Jobs
|
|
||||||
|
|
||||||
from selfprivacy_api.services.flake_service_manager import FlakeServiceManager
|
|
||||||
|
|
||||||
|
|
||||||
class UpdateServicesFlakeList(Migration):
|
|
||||||
"""Check if all required services are in the flake list"""
|
|
||||||
|
|
||||||
def get_migration_name(self):
|
|
||||||
return "update_services_flake_list"
|
|
||||||
|
|
||||||
def get_migration_description(self):
|
|
||||||
return "Check if all required services are in the flake list"
|
|
||||||
|
|
||||||
def is_migration_needed(self):
|
|
||||||
with FlakeServiceManager() as manager:
|
|
||||||
if "roundcube" not in manager.services:
|
|
||||||
return True
|
|
||||||
|
|
||||||
def migrate(self):
|
|
||||||
with FlakeServiceManager() as manager:
|
|
||||||
if "roundcube" not in manager.services:
|
|
||||||
manager.services[
|
|
||||||
"roundcube"
|
|
||||||
] = "git+https://git.selfprivacy.org/SelfPrivacy/selfprivacy-nixos-config.git?ref=flakes&dir=sp-modules/roundcube"
|
|
|
@ -15,10 +15,10 @@ from selfprivacy_api.utils import ReadUserData, UserDataFiles
|
||||||
class WriteTokenToRedis(Migration):
|
class WriteTokenToRedis(Migration):
|
||||||
"""Load Json tokens into Redis"""
|
"""Load Json tokens into Redis"""
|
||||||
|
|
||||||
def get_migration_name(self):
|
def get_migration_name(self) -> str:
|
||||||
return "write_token_to_redis"
|
return "write_token_to_redis"
|
||||||
|
|
||||||
def get_migration_description(self):
|
def get_migration_description(self) -> str:
|
||||||
return "Loads the initial token into redis token storage"
|
return "Loads the initial token into redis token storage"
|
||||||
|
|
||||||
def is_repo_empty(self, repo: AbstractTokensRepository) -> bool:
|
def is_repo_empty(self, repo: AbstractTokensRepository) -> bool:
|
||||||
|
@ -38,7 +38,7 @@ class WriteTokenToRedis(Migration):
|
||||||
print(e)
|
print(e)
|
||||||
return None
|
return None
|
||||||
|
|
||||||
def is_migration_needed(self):
|
def is_migration_needed(self) -> bool:
|
||||||
try:
|
try:
|
||||||
if self.get_token_from_json() is not None and self.is_repo_empty(
|
if self.get_token_from_json() is not None and self.is_repo_empty(
|
||||||
RedisTokensRepository()
|
RedisTokensRepository()
|
||||||
|
@ -47,8 +47,9 @@ class WriteTokenToRedis(Migration):
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
print(e)
|
print(e)
|
||||||
return False
|
return False
|
||||||
|
return False
|
||||||
|
|
||||||
def migrate(self):
|
def migrate(self) -> None:
|
||||||
# Write info about providers to userdata.json
|
# Write info about providers to userdata.json
|
||||||
try:
|
try:
|
||||||
token = self.get_token_from_json()
|
token = self.get_token_from_json()
|
||||||
|
|
|
@ -37,14 +37,14 @@ class Bitwarden(Service):
|
||||||
def get_user() -> str:
|
def get_user() -> str:
|
||||||
return "vaultwarden"
|
return "vaultwarden"
|
||||||
|
|
||||||
@staticmethod
|
@classmethod
|
||||||
def get_url() -> Optional[str]:
|
def get_url(cls) -> Optional[str]:
|
||||||
"""Return service url."""
|
"""Return service url."""
|
||||||
domain = get_domain()
|
domain = get_domain()
|
||||||
return f"https://password.{domain}"
|
return f"https://password.{domain}"
|
||||||
|
|
||||||
@staticmethod
|
@classmethod
|
||||||
def get_subdomain() -> Optional[str]:
|
def get_subdomain(cls) -> Optional[str]:
|
||||||
return "password"
|
return "password"
|
||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
|
|
|
@ -33,14 +33,14 @@ class Gitea(Service):
|
||||||
"""Read SVG icon from file and return it as base64 encoded string."""
|
"""Read SVG icon from file and return it as base64 encoded string."""
|
||||||
return base64.b64encode(GITEA_ICON.encode("utf-8")).decode("utf-8")
|
return base64.b64encode(GITEA_ICON.encode("utf-8")).decode("utf-8")
|
||||||
|
|
||||||
@staticmethod
|
@classmethod
|
||||||
def get_url() -> Optional[str]:
|
def get_url(cls) -> Optional[str]:
|
||||||
"""Return service url."""
|
"""Return service url."""
|
||||||
domain = get_domain()
|
domain = get_domain()
|
||||||
return f"https://git.{domain}"
|
return f"https://git.{domain}"
|
||||||
|
|
||||||
@staticmethod
|
@classmethod
|
||||||
def get_subdomain() -> Optional[str]:
|
def get_subdomain(cls) -> Optional[str]:
|
||||||
return "git"
|
return "git"
|
||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
|
|
|
@ -36,14 +36,14 @@ class JitsiMeet(Service):
|
||||||
"""Read SVG icon from file and return it as base64 encoded string."""
|
"""Read SVG icon from file and return it as base64 encoded string."""
|
||||||
return base64.b64encode(JITSI_ICON.encode("utf-8")).decode("utf-8")
|
return base64.b64encode(JITSI_ICON.encode("utf-8")).decode("utf-8")
|
||||||
|
|
||||||
@staticmethod
|
@classmethod
|
||||||
def get_url() -> Optional[str]:
|
def get_url(cls) -> Optional[str]:
|
||||||
"""Return service url."""
|
"""Return service url."""
|
||||||
domain = get_domain()
|
domain = get_domain()
|
||||||
return f"https://meet.{domain}"
|
return f"https://meet.{domain}"
|
||||||
|
|
||||||
@staticmethod
|
@classmethod
|
||||||
def get_subdomain() -> Optional[str]:
|
def get_subdomain(cls) -> Optional[str]:
|
||||||
return "meet"
|
return "meet"
|
||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
|
|
|
@ -35,13 +35,13 @@ class MailServer(Service):
|
||||||
def get_user() -> str:
|
def get_user() -> str:
|
||||||
return "virtualMail"
|
return "virtualMail"
|
||||||
|
|
||||||
@staticmethod
|
@classmethod
|
||||||
def get_url() -> Optional[str]:
|
def get_url(cls) -> Optional[str]:
|
||||||
"""Return service url."""
|
"""Return service url."""
|
||||||
return None
|
return None
|
||||||
|
|
||||||
@staticmethod
|
@classmethod
|
||||||
def get_subdomain() -> Optional[str]:
|
def get_subdomain(cls) -> Optional[str]:
|
||||||
return None
|
return None
|
||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
|
|
|
@ -4,7 +4,6 @@ import subprocess
|
||||||
from typing import Optional, List
|
from typing import Optional, List
|
||||||
|
|
||||||
from selfprivacy_api.utils import get_domain
|
from selfprivacy_api.utils import get_domain
|
||||||
from selfprivacy_api.jobs import Job, Jobs
|
|
||||||
|
|
||||||
from selfprivacy_api.utils.systemd import get_service_status
|
from selfprivacy_api.utils.systemd import get_service_status
|
||||||
from selfprivacy_api.services.service import Service, ServiceStatus
|
from selfprivacy_api.services.service import Service, ServiceStatus
|
||||||
|
@ -35,14 +34,14 @@ class Nextcloud(Service):
|
||||||
"""Read SVG icon from file and return it as base64 encoded string."""
|
"""Read SVG icon from file and return it as base64 encoded string."""
|
||||||
return base64.b64encode(NEXTCLOUD_ICON.encode("utf-8")).decode("utf-8")
|
return base64.b64encode(NEXTCLOUD_ICON.encode("utf-8")).decode("utf-8")
|
||||||
|
|
||||||
@staticmethod
|
@classmethod
|
||||||
def get_url() -> Optional[str]:
|
def get_url(cls) -> Optional[str]:
|
||||||
"""Return service url."""
|
"""Return service url."""
|
||||||
domain = get_domain()
|
domain = get_domain()
|
||||||
return f"https://cloud.{domain}"
|
return f"https://cloud.{domain}"
|
||||||
|
|
||||||
@staticmethod
|
@classmethod
|
||||||
def get_subdomain() -> Optional[str]:
|
def get_subdomain(cls) -> Optional[str]:
|
||||||
return "cloud"
|
return "cloud"
|
||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
|
|
|
@ -28,13 +28,13 @@ class Ocserv(Service):
|
||||||
def get_svg_icon() -> str:
|
def get_svg_icon() -> str:
|
||||||
return base64.b64encode(OCSERV_ICON.encode("utf-8")).decode("utf-8")
|
return base64.b64encode(OCSERV_ICON.encode("utf-8")).decode("utf-8")
|
||||||
|
|
||||||
@staticmethod
|
@classmethod
|
||||||
def get_url() -> typing.Optional[str]:
|
def get_url(cls) -> typing.Optional[str]:
|
||||||
"""Return service url."""
|
"""Return service url."""
|
||||||
return None
|
return None
|
||||||
|
|
||||||
@staticmethod
|
@classmethod
|
||||||
def get_subdomain() -> typing.Optional[str]:
|
def get_subdomain(cls) -> typing.Optional[str]:
|
||||||
return "vpn"
|
return "vpn"
|
||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
|
|
|
@ -31,14 +31,14 @@ class Pleroma(Service):
|
||||||
def get_svg_icon() -> str:
|
def get_svg_icon() -> str:
|
||||||
return base64.b64encode(PLEROMA_ICON.encode("utf-8")).decode("utf-8")
|
return base64.b64encode(PLEROMA_ICON.encode("utf-8")).decode("utf-8")
|
||||||
|
|
||||||
@staticmethod
|
@classmethod
|
||||||
def get_url() -> Optional[str]:
|
def get_url(cls) -> Optional[str]:
|
||||||
"""Return service url."""
|
"""Return service url."""
|
||||||
domain = get_domain()
|
domain = get_domain()
|
||||||
return f"https://social.{domain}"
|
return f"https://social.{domain}"
|
||||||
|
|
||||||
@staticmethod
|
@classmethod
|
||||||
def get_subdomain() -> Optional[str]:
|
def get_subdomain(cls) -> Optional[str]:
|
||||||
return "social"
|
return "social"
|
||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
|
|
|
@ -2,14 +2,14 @@
|
||||||
|
|
||||||
import base64
|
import base64
|
||||||
import subprocess
|
import subprocess
|
||||||
from typing import Optional, List
|
from typing import List, Optional
|
||||||
|
|
||||||
from selfprivacy_api.jobs import Job
|
from selfprivacy_api.jobs import Job
|
||||||
from selfprivacy_api.utils.systemd import (
|
from selfprivacy_api.utils.systemd import (
|
||||||
get_service_status_from_several_units,
|
get_service_status_from_several_units,
|
||||||
)
|
)
|
||||||
from selfprivacy_api.services.service import Service, ServiceStatus
|
from selfprivacy_api.services.service import Service, ServiceStatus
|
||||||
from selfprivacy_api.utils import get_domain
|
from selfprivacy_api.utils import ReadUserData, get_domain
|
||||||
from selfprivacy_api.utils.block_devices import BlockDevice
|
from selfprivacy_api.utils.block_devices import BlockDevice
|
||||||
from selfprivacy_api.services.roundcube.icon import ROUNDCUBE_ICON
|
from selfprivacy_api.services.roundcube.icon import ROUNDCUBE_ICON
|
||||||
|
|
||||||
|
@ -37,20 +37,20 @@ class Roundcube(Service):
|
||||||
"""Read SVG icon from file and return it as base64 encoded string."""
|
"""Read SVG icon from file and return it as base64 encoded string."""
|
||||||
return base64.b64encode(ROUNDCUBE_ICON.encode("utf-8")).decode("utf-8")
|
return base64.b64encode(ROUNDCUBE_ICON.encode("utf-8")).decode("utf-8")
|
||||||
|
|
||||||
@staticmethod
|
@classmethod
|
||||||
def get_url() -> Optional[str]:
|
def get_url(cls) -> Optional[str]:
|
||||||
"""Return service url."""
|
"""Return service url."""
|
||||||
domain = get_domain()
|
domain = get_domain()
|
||||||
subdomain = get_subdomain()
|
subdomain = cls.get_subdomain()
|
||||||
return f"https://{subdomain}.{domain}"
|
return f"https://{subdomain}.{domain}"
|
||||||
|
|
||||||
@staticmethod
|
@classmethod
|
||||||
def get_subdomain() -> Optional[str]:
|
def get_subdomain(cls) -> Optional[str]:
|
||||||
with ReadUserData() as data:
|
with ReadUserData() as data:
|
||||||
if "roundcube" in data["modules"]:
|
if "roundcube" in data["modules"]:
|
||||||
return data["modules"]["roundcube"]["subdomain"]
|
return data["modules"]["roundcube"]["subdomain"]
|
||||||
|
|
||||||
return "webmail"
|
return "roundcube"
|
||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
def is_movable() -> bool:
|
def is_movable() -> bool:
|
||||||
|
|
|
@ -65,17 +65,17 @@ class Service(ABC):
|
||||||
"""
|
"""
|
||||||
pass
|
pass
|
||||||
|
|
||||||
@staticmethod
|
@classmethod
|
||||||
@abstractmethod
|
@abstractmethod
|
||||||
def get_url() -> Optional[str]:
|
def get_url(cls) -> Optional[str]:
|
||||||
"""
|
"""
|
||||||
The url of the service if it is accessible from the internet browser.
|
The url of the service if it is accessible from the internet browser.
|
||||||
"""
|
"""
|
||||||
pass
|
pass
|
||||||
|
|
||||||
@staticmethod
|
@classmethod
|
||||||
@abstractmethod
|
@abstractmethod
|
||||||
def get_subdomain() -> Optional[str]:
|
def get_subdomain(cls) -> Optional[str]:
|
||||||
"""
|
"""
|
||||||
The assigned primary subdomain for this service.
|
The assigned primary subdomain for this service.
|
||||||
"""
|
"""
|
||||||
|
|
|
@ -57,14 +57,14 @@ class DummyService(Service):
|
||||||
# return ""
|
# return ""
|
||||||
return base64.b64encode(BITWARDEN_ICON.encode("utf-8")).decode("utf-8")
|
return base64.b64encode(BITWARDEN_ICON.encode("utf-8")).decode("utf-8")
|
||||||
|
|
||||||
@staticmethod
|
@classmethod
|
||||||
def get_url() -> typing.Optional[str]:
|
def get_url(cls) -> typing.Optional[str]:
|
||||||
"""Return service url."""
|
"""Return service url."""
|
||||||
domain = "test.com"
|
domain = "test.com"
|
||||||
return f"https://password.{domain}"
|
return f"https://password.{domain}"
|
||||||
|
|
||||||
@staticmethod
|
@classmethod
|
||||||
def get_subdomain() -> typing.Optional[str]:
|
def get_subdomain(cls) -> typing.Optional[str]:
|
||||||
return "password"
|
return "password"
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
|
|
|
@ -62,6 +62,9 @@
|
||||||
"simple-nixos-mailserver": {
|
"simple-nixos-mailserver": {
|
||||||
"enable": true,
|
"enable": true,
|
||||||
"location": "sdb"
|
"location": "sdb"
|
||||||
|
},
|
||||||
|
"roundcube": {
|
||||||
|
"enable": true
|
||||||
}
|
}
|
||||||
},
|
},
|
||||||
"volumes": [
|
"volumes": [
|
||||||
|
|
Loading…
Reference in a new issue