refactor(config): rename env to config and use pydantic manage config

.gitignore

@@ -1,5 +1,6 @@
 .idea/
 .vscode/
+config.toml
 
 # Created by https://www.toptal.com/developers/gitignore/api/python
 # Edit at https://www.toptal.com/developers/gitignore?templates=python

config.example.toml

@@ -0,0 +1,11 @@
+[owner]
+name = "podmaker"
+email = "[email protected]"
+
+[s3]
+access_key = "123"
+access_secret = "456"
+bucket = "podmake"
+endpoint = "https://s3.amazonaws.com"
+# must be public-read
+cdn_prefix = "https://s3.amazonaws.com"

podmaker/config.py

@@ -0,0 +1,35 @@
+from __future__ import annotations
+
+import sys
+from pathlib import PurePath
+
+from pydantic import BaseModel, EmailStr, Field, HttpUrl
+
+if sys.version_info >= (3, 11):
+    import tomllib as toml
+else:
+    import tomlkit as toml
+
+
+class OwnerConfig(BaseModel):
+    name: str = Field(min_length=1, frozen=True)
+    email: EmailStr = Field(frozen=True)
+
+
+class S3Config(BaseModel):
+    access_key: str = Field(min_length=1, frozen=True)
+    access_secret: str = Field(min_length=1, frozen=True)
+    bucket: str = Field(min_length=1, frozen=True)
+    endpoint: HttpUrl = Field(frozen=True)
+    cdn_prefix: HttpUrl = Field(frozen=True)
+
+
+class PMConfig(BaseModel):
+    owner: OwnerConfig = Field(frozen=True)
+    s3: S3Config = Field(frozen=True)
+
+    @classmethod
+    def from_file(cls, path: PurePath) -> PMConfig:
+        with open(path, 'rb') as f:
+            data = toml.load(f)
+        return cls(**data)

podmaker/parser/youtube.py

@@ -10,7 +10,7 @@
 
 import yt_dlp
 
-from podmaker.env import PMEnv
+from podmaker.config import PMConfig
 from podmaker.parser import Parser
 from podmaker.rss import Enclosure, Episode, Owner, Podcast, Resource
 from podmaker.storage import Storage
@@ -33,10 +33,10 @@ def error(self, msg: Any) -> None:
 
 
 class YouTube(Parser):
-    def __init__(self, storage: Storage, env: PMEnv):
+    def __init__(self, storage: Storage, config: PMConfig):
         self.storage = storage
         self.ydl_opts = {'logger': NoneLogger()}
-        self.env = env
+        self.config = config
 
     def fetch(self, uri: ParseResult) -> Podcast:
         if uri.path == "/playlist":
@@ -53,7 +53,7 @@ def fetch_playlist(self, url: str) -> Podcast:
                 title=playlist['title'],
                 image=PlaylistThumbnail(playlist['thumbnails']),
                 description=playlist['description'],
-                owner=Owner(name=self.env.owner.name, email=self.env.owner.email),
+                owner=Owner(name=self.config.owner.name, email=self.config.owner.email),
                 author=playlist['uploader'],
                 categories=playlist.get('tags', []),
             )

podmaker/storage/s3.py

@@ -9,7 +9,7 @@
 import boto3
 from botocore.exceptions import ClientError
 
-from podmaker.env import PMEnv
+from podmaker.config import PMConfig
 from podmaker.storage import ObjectInfo, Storage
 
 logger = logging.getLogger(__name__)
@@ -18,13 +18,12 @@
 class S3(Storage):
     _md5_chunk_size = 1024 * 1024
 
-    def __init__(self, env: PMEnv):
-        s3_env = env.s3
+    def __init__(self, config: PMConfig):
         self.s3 = boto3.resource(
-            's3', endpoint_url=s3_env.endpoint, aws_access_key_id=s3_env.access_key,
-            aws_secret_access_key=s3_env.access_secret)
-        self.bucket = self.s3.Bucket(s3_env.bucket)
-        self.cdn_prefix = s3_env.cdn_prefix
+            's3', endpoint_url=str(config.s3.endpoint), aws_access_key_id=config.s3.access_key,
+            aws_secret_access_key=config.s3.access_secret)
+        self.bucket = self.s3.Bucket(config.s3.bucket)
+        self.cdn_prefix = str(config.s3.cdn_prefix)
 
     def _calculate_md5(self, data: IO[AnyStr]) -> str:
         md5 = hashlib.md5()