Compare commits
10 Commits
renovate/c
...
main
| Author | SHA1 | Date | |
|---|---|---|---|
| 0786843e71 | |||
|
d27ace922d
|
|||
| d5bf118df9 | |||
|
d60e83a019
|
|||
| 85dfad97c2 | |||
|
02617e1033
|
|||
| e91531f4b1 | |||
|
fe3bcf3980
|
|||
| 451f69935f | |||
|
fc9ded68b3
|
@@ -2,82 +2,82 @@ on:
|
||||
workflow_dispatch:
|
||||
inputs:
|
||||
github_release:
|
||||
description: 'Create Gitea Release'
|
||||
description: "Create Gitea Release"
|
||||
default: true
|
||||
type: boolean
|
||||
bump:
|
||||
description: 'Bump type'
|
||||
description: "Bump type"
|
||||
required: true
|
||||
default: 'patch'
|
||||
default: "patch"
|
||||
type: choice
|
||||
options:
|
||||
- 'major'
|
||||
- 'minor'
|
||||
- 'patch'
|
||||
|
||||
- "major"
|
||||
- "minor"
|
||||
- "patch"
|
||||
|
||||
jobs:
|
||||
build:
|
||||
runs-on: ubuntu-latest
|
||||
steps:
|
||||
- name: Checkout code
|
||||
uses: actions/checkout@master
|
||||
with:
|
||||
fetch-depth: 0
|
||||
fetch-tags: true
|
||||
- name: Install uv
|
||||
uses: astral-sh/setup-uv@v5
|
||||
- name: Set up Python
|
||||
run: uv python install
|
||||
with:
|
||||
python-version-file: "pyproject.toml"
|
||||
- name: Set Git identity
|
||||
run: |
|
||||
git config user.name "Gitea CI"
|
||||
git config user.email "ci@git.theprivateserver.de"
|
||||
- name: Bump version
|
||||
id: bump
|
||||
run: |
|
||||
uv tool install bump-my-version
|
||||
uv tool run bump-my-version bump ${{ github.event.inputs.bump }}
|
||||
# echo the version to github env, the version is shown by using uv tool run bump-my-version show current_version
|
||||
echo "VERSION<<EOF" >> $GITHUB_ENV
|
||||
echo "$(uv tool run bump-my-version show current_version)" >> $GITHUB_ENV
|
||||
echo "EOF" >> $GITHUB_ENV
|
||||
- name: Push changes
|
||||
uses: ad-m/github-push-action@master
|
||||
with:
|
||||
github_token: ${{ secrets.GITHUB_TOKEN }}
|
||||
branch: ${{ github.ref }}
|
||||
|
||||
- name: Build Changelog
|
||||
id: build_changelog
|
||||
uses: https://github.com/mikepenz/release-changelog-builder-action@v5
|
||||
with:
|
||||
platform: "gitea"
|
||||
baseURL: "http://gitea:3000"
|
||||
configuration: ".gitea/changelog-config.json"
|
||||
env:
|
||||
GITHUB_TOKEN: ${{ secrets.GITEA_TOKEN }}
|
||||
- name: Build package
|
||||
run: uv build
|
||||
- name: Publish package
|
||||
env:
|
||||
USERNAME: ${{ github.repository_owner }}
|
||||
run: uv publish --publish-url https://git.theprivateserver.de/api/packages/$USERNAME/pypi/ -t ${{ secrets.TOKEN }}
|
||||
|
||||
- name: Checkout code
|
||||
uses: actions/checkout@v5
|
||||
with:
|
||||
fetch-depth: 0
|
||||
fetch-tags: true
|
||||
- name: Install uv
|
||||
uses: astral-sh/setup-uv@v5
|
||||
- name: Set up Python
|
||||
run: uv python install
|
||||
with:
|
||||
python-version-file: "pyproject.toml"
|
||||
- name: Set Git identity
|
||||
run: |
|
||||
git config user.name "Gitea CI"
|
||||
git config user.email "ci@git.theprivateserver.de"
|
||||
- name: Bump version
|
||||
id: bump
|
||||
run: |
|
||||
uv tool install bump-my-version
|
||||
uv tool run bump-my-version bump ${{ github.event.inputs.bump }}
|
||||
# echo the version to github env, the version is shown by using uv tool run bump-my-version show current_version
|
||||
echo "VERSION<<EOF" >> $GITHUB_ENV
|
||||
echo "$(uv tool run bump-my-version show current_version)" >> $GITHUB_ENV
|
||||
echo "EOF" >> $GITHUB_ENV
|
||||
- name: Push changes
|
||||
uses: ad-m/github-push-action@master
|
||||
with:
|
||||
github_token: ${{ secrets.GITHUB_TOKEN }}
|
||||
branch: ${{ github.ref }}
|
||||
|
||||
- name: Create release
|
||||
id: create_release
|
||||
if: ${{ github.event.inputs.github_release == 'true' }}
|
||||
uses: softprops/action-gh-release@master
|
||||
with:
|
||||
tag_name: ${{ env.VERSION }}
|
||||
release_name: Release ${{ env.VERSION }}
|
||||
body: ${{steps.build_changelog.outputs.changelog}}
|
||||
draft: false
|
||||
prerelease: false
|
||||
make_latest: true
|
||||
files: |
|
||||
dist/*
|
||||
env:
|
||||
GITHUB_TOKEN: ${{ secrets.TOKEN }}
|
||||
- name: Build Changelog
|
||||
id: build_changelog
|
||||
continue-on-error: true
|
||||
uses: https://github.com/mikepenz/release-changelog-builder-action@v5
|
||||
with:
|
||||
platform: "gitea"
|
||||
baseURL: "http://192.168.178.110:3000"
|
||||
configuration: ".gitea/changelog-config.json"
|
||||
env:
|
||||
GITHUB_TOKEN: ${{ secrets.GITEA_TOKEN }}
|
||||
- name: Build package
|
||||
run: uv build
|
||||
- name: Publish package
|
||||
env:
|
||||
USERNAME: ${{ github.repository_owner }}
|
||||
run: uv publish --publish-url https://git.theprivateserver.de/api/packages/$USERNAME/pypi/ -t ${{ secrets.TOKEN }}
|
||||
|
||||
- name: Create release
|
||||
id: create_release
|
||||
if: ${{ github.event.inputs.github_release == 'true' }}
|
||||
uses: softprops/action-gh-release@master
|
||||
with:
|
||||
tag_name: v${{ env.VERSION }}
|
||||
release_name: Release ${{ env.VERSION }}
|
||||
body: ${{ steps.build_changelog.outcome == 'success' && steps.build_changelog.outputs.changelog || 'No changelog available' }}
|
||||
draft: false
|
||||
prerelease: false
|
||||
make_latest: true
|
||||
files: |
|
||||
dist/*
|
||||
env:
|
||||
GITHUB_TOKEN: ${{ secrets.TOKEN }}
|
||||
|
||||
7
.vscode/settings.json
vendored
Normal file
7
.vscode/settings.json
vendored
Normal file
@@ -0,0 +1,7 @@
|
||||
{
|
||||
"python.testing.pytestArgs": [
|
||||
"test"
|
||||
],
|
||||
"python.testing.unittestEnabled": false,
|
||||
"python.testing.pytestEnabled": true
|
||||
}
|
||||
@@ -1,14 +1,58 @@
|
||||
[project]
|
||||
name = "comicvineapi"
|
||||
version = "0.1.0"
|
||||
version = "0.0.0"
|
||||
description = "Add your description here"
|
||||
readme = "README.md"
|
||||
authors = [
|
||||
{ name = "WorldTeacher", email = "coding_contact@pm.me" }
|
||||
]
|
||||
requires-python = ">=3.13"
|
||||
dependencies = []
|
||||
dependencies = [
|
||||
"httpx>=0.28.1",
|
||||
"httpx-ratelimiter>=0.0.6",
|
||||
"httpx-retries>=0.4.0",
|
||||
"pyrate-limiter>=3.1.1",
|
||||
]
|
||||
|
||||
[build-system]
|
||||
requires = ["hatchling"]
|
||||
build-backend = "hatchling.build"
|
||||
|
||||
[tool.pytest.ini_options]
|
||||
testpaths = ["test"]
|
||||
addopts = "--cov=src --cov-report=term-missing"
|
||||
[tool.coverage.run]
|
||||
omit = ["main.py", "test.py", "test/*", "__init__.py"]
|
||||
|
||||
|
||||
[tool.bumpversion]
|
||||
current_version = "0.0.0"
|
||||
parse = "(?P<major>\\d+)\\.(?P<minor>\\d+)\\.(?P<patch>\\d+)"
|
||||
serialize = ["{major}.{minor}.{patch}"]
|
||||
search = "{current_version}"
|
||||
replace = "{new_version}"
|
||||
regex = false
|
||||
ignore_missing_version = false
|
||||
ignore_missing_files = false
|
||||
tag = false
|
||||
sign_tags = false
|
||||
tag_name = "v{new_version}"
|
||||
tag_message = "Bump version: {current_version} → {new_version}"
|
||||
allow_dirty = false
|
||||
commit = false
|
||||
message = "Bump version: {current_version} → {new_version}"
|
||||
moveable_tags = []
|
||||
commit_args = ""
|
||||
setup_hooks = []
|
||||
pre_commit_hooks = []
|
||||
post_commit_hooks = []
|
||||
|
||||
|
||||
[dependency-groups]
|
||||
dev = [
|
||||
"python-dotenv>=1.1.1",
|
||||
]
|
||||
test = [
|
||||
"pytest>=8.4.1",
|
||||
"pytest-cov>=6.2.1",
|
||||
]
|
||||
|
||||
@@ -1,2 +1,5 @@
|
||||
def hello() -> str:
|
||||
return "Hello from comicvineapi!"
|
||||
__version__ = "0.1.5"
|
||||
__all__ = ["ComicVineAPI", "Cache"]
|
||||
from .api import ComicVineAPI
|
||||
from .cache import Cache
|
||||
from .schemas.api_classes import *
|
||||
376
src/comicvineapi/api.py
Normal file
376
src/comicvineapi/api.py
Normal file
@@ -0,0 +1,376 @@
|
||||
import sys
|
||||
from datetime import datetime
|
||||
from typing import Optional, Dict, Any, Union
|
||||
|
||||
import httpx
|
||||
from httpx_retries import RetryTransport, Retry
|
||||
from httpx_ratelimiter import LimiterTransport
|
||||
from pyrate_limiter import Rate, Duration
|
||||
from loguru import logger
|
||||
from urllib.parse import urlencode
|
||||
|
||||
from .schemas.response import ComicVineResponse
|
||||
from .schemas.api_classes import (
|
||||
Character,
|
||||
Characters,
|
||||
Chat,
|
||||
Concept,
|
||||
Episode,
|
||||
Episodes,
|
||||
Location,
|
||||
Movie,
|
||||
Object,
|
||||
Person,
|
||||
People,
|
||||
Power,
|
||||
Promo,
|
||||
Publisher,
|
||||
Series,
|
||||
StoryArc,
|
||||
StoryArcs,
|
||||
Team,
|
||||
Teams,
|
||||
Type,
|
||||
Video,
|
||||
VideoCategory,
|
||||
VideoType,
|
||||
Volume,
|
||||
Volumes,
|
||||
Issue,
|
||||
Issues,
|
||||
)
|
||||
from .cache import Cache
|
||||
from comicvineapi import __version__
|
||||
|
||||
log = logger
|
||||
log.remove()
|
||||
log.add(
|
||||
f"logs/{datetime.now().strftime('%Y-%m-%d')}.log",
|
||||
rotation="1 day",
|
||||
compression="zip",
|
||||
)
|
||||
log.add("logs/api.log", compression="zip", rotation="50 MB")
|
||||
# log.add(sys.stderr)
|
||||
|
||||
|
||||
class ComicVineAPI:
|
||||
def __init__(
|
||||
self,
|
||||
api_key: str,
|
||||
cache: Optional[Cache] = None,
|
||||
*,
|
||||
# match old @limits(calls=20, period=60)
|
||||
rate_per_minute: int = 20,
|
||||
# sensible retry defaults; 429, 502–504 covered by default
|
||||
retries_total: int = 5,
|
||||
backoff_factor: float = 0.5,
|
||||
timeout: Union[float, httpx.Timeout] = httpx.Timeout(10.0, connect=5.0),
|
||||
follow_redirects: bool = True,
|
||||
):
|
||||
self.api_key = api_key
|
||||
self.cache = cache
|
||||
self.api_url = "https://comicvine.gamespot.com/api/"
|
||||
self.userAgent = f"ComicVineAPI/{__version__}"
|
||||
|
||||
# Build transport chain: limiter (rate) -> retry wrapper
|
||||
limiter = LimiterTransport(rates=[Rate(rate_per_minute, Duration.MINUTE)])
|
||||
retry_cfg = Retry(total=retries_total, backoff_factor=backoff_factor)
|
||||
transport = RetryTransport(transport=limiter, retry=retry_cfg)
|
||||
|
||||
self.client = httpx.Client(
|
||||
headers={"User-Agent": self.userAgent},
|
||||
timeout=timeout,
|
||||
follow_redirects=follow_redirects,
|
||||
transport=transport,
|
||||
)
|
||||
|
||||
if not self._test_reachable():
|
||||
log.warning(f"Could not connect to {self.api_url}")
|
||||
self.client.close()
|
||||
raise ConnectionError(f"Could not connect to {self.api_url}")
|
||||
|
||||
log.info("ComicVineAPI initialized")
|
||||
|
||||
def _test_reachable(self) -> bool:
|
||||
try:
|
||||
r = self.client.get(self.api_url, timeout=5.0)
|
||||
r.raise_for_status()
|
||||
return True
|
||||
except httpx.RequestError:
|
||||
return False
|
||||
|
||||
def handle_kwargs(self, kwargs: Dict[str, Any]) -> Dict[str, str]:
|
||||
"""
|
||||
Build ComicVine 'filter' from kwargs (key:value pairs).
|
||||
Non-filter params like sort/offset can be added later by callers.
|
||||
"""
|
||||
params: Dict[str, str] = {}
|
||||
parts = []
|
||||
for key, value in kwargs.items():
|
||||
log.debug(f"Handling key: {key} with value: {value}")
|
||||
if value is not None:
|
||||
parts.append(f"{key}:{value}")
|
||||
if parts:
|
||||
params["filter"] = ",".join(parts)
|
||||
log.debug(f"Handling kwargs -> params: {params}")
|
||||
return params
|
||||
|
||||
@log.catch
|
||||
def get_request(self, url: str, filters: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
|
||||
filters = dict(filters or {})
|
||||
filters["api_key"] = self.api_key
|
||||
filters["format"] = "json"
|
||||
|
||||
cache_params = f"?{urlencode({k: str(filters[k]) for k in sorted(filters)})}"
|
||||
cache_query = f"{self.api_url}{url}{cache_params}".replace(self.api_key, "API_KEY")
|
||||
|
||||
if self.cache:
|
||||
log.debug(f"Checking cache for {cache_query}")
|
||||
cached_response = self.cache.get_response(cache_query)
|
||||
if cached_response:
|
||||
return cached_response
|
||||
|
||||
response = self.client.get(self.api_url + url, params=filters)
|
||||
response.raise_for_status()
|
||||
res = response.json()
|
||||
if res is None:
|
||||
raise ValueError(f"No response from {url}")
|
||||
|
||||
if self.cache:
|
||||
self.cache.insert_response(cache_query, res)
|
||||
return res
|
||||
|
||||
def get_all_request(
|
||||
self,
|
||||
url: str,
|
||||
limit: int,
|
||||
filters: Optional[Dict[str, Union[str, int, float]]] = None,
|
||||
) -> Dict[str, Any]:
|
||||
filters = dict(filters or {})
|
||||
filters["api_key"] = self.api_key
|
||||
filters["format"] = "json"
|
||||
|
||||
params = f"?{urlencode({k: str(filters[k]) for k in sorted(filters)})}"
|
||||
all_url = f"{self.api_url}{url}{params}".replace(self.api_key, "API_KEY")
|
||||
|
||||
if self.cache:
|
||||
cached_response = self.cache.get_response(all_url)
|
||||
if cached_response:
|
||||
log.info(f"Got cached response for {all_url}")
|
||||
return cached_response
|
||||
|
||||
response = self.get_request(url, filters)
|
||||
|
||||
# keep fetching until we hit server total or client limit
|
||||
def _cap() -> int:
|
||||
return min(response.get("number_of_total_results", 0) or 0, limit)
|
||||
|
||||
while response.get("results") and len(response["results"]) < _cap():
|
||||
filters["offset"] = len(response["results"])
|
||||
log.debug(f"Getting next page with offset {filters['offset']}")
|
||||
next_response = self.get_request(url, filters)
|
||||
response["results"].extend(next_response.get("results", []))
|
||||
|
||||
if self.cache:
|
||||
self.cache.insert_response(all_url, response)
|
||||
return response
|
||||
|
||||
# ----- Endpoints -----
|
||||
|
||||
def get_character(self, id: int, **kwargs) -> Character:
|
||||
url = f"character/4005-{id}"
|
||||
filters = self.handle_kwargs(kwargs)
|
||||
response = self.get_request(url, filters)
|
||||
return ComicVineResponse(**response).handle_result(Character)
|
||||
|
||||
def get_characters(self, all: bool = False, limit: int = 1000, **kwargs) -> list[Characters]:
|
||||
url = "characters/"
|
||||
filters = self.handle_kwargs(kwargs)
|
||||
response = self.get_all_request(url, limit, filters) if all else self.get_request(url, filters)
|
||||
return ComicVineResponse(**response).handle_result(Characters)
|
||||
|
||||
def get_chat(self, id: int) -> Chat:
|
||||
return self.get_chats(id=id)[0]
|
||||
|
||||
def get_chats(self, all: bool = False, limit: int = 1000, **kwargs) -> list[Chat]:
|
||||
url = "chats/"
|
||||
filters = self.handle_kwargs(kwargs)
|
||||
response = self.get_all_request(url, limit, filters) if all else self.get_request(url, filters)
|
||||
return ComicVineResponse(**response).handle_result(Chat)
|
||||
|
||||
def get_concept(self, id: int) -> Concept:
|
||||
url = f"concept/4015-{id}"
|
||||
return ComicVineResponse(**self.get_request(url)).handle_result(Concept)
|
||||
|
||||
def get_concepts(self, all: bool = False, limit: int = 1000, **kwargs) -> list[Concept]:
|
||||
url = "concepts/"
|
||||
filters = self.handle_kwargs(kwargs)
|
||||
response = self.get_all_request(url, limit, filters) if all else self.get_request(url, filters)
|
||||
return ComicVineResponse(**response).handle_result(Concept)
|
||||
|
||||
def get_episode(self, id: int) -> Episode:
|
||||
url = f"episode/4070-{id}"
|
||||
return ComicVineResponse(**self.get_request(url)).handle_result(Episode)
|
||||
|
||||
def get_episodes(self, all: bool = False, limit: int = 1000, **kwargs) -> list[Episodes]:
|
||||
url = "episodes/"
|
||||
filters = self.handle_kwargs(kwargs)
|
||||
response = self.get_all_request(url, limit, filters) if all else self.get_request(url, filters)
|
||||
return ComicVineResponse(**response).handle_result(Episodes)
|
||||
|
||||
def get_issue(self, id: int) -> Issue:
|
||||
url = f"issue/4000-{id}"
|
||||
return ComicVineResponse(**self.get_request(url)).handle_result(Issue)
|
||||
|
||||
def get_issues(self, all: bool = False, limit: int = 1000, **kwargs) -> list[Issues]:
|
||||
url = "issues/"
|
||||
filters = self.handle_kwargs(kwargs)
|
||||
response = self.get_all_request(url, limit, filters) if all else self.get_request(url, filters)
|
||||
return ComicVineResponse(**response).handle_result(Issues)
|
||||
|
||||
def get_location(self, id: int) -> Location:
|
||||
url = f"location/4020-{id}"
|
||||
return ComicVineResponse(**self.get_request(url)).handle_result(Location)
|
||||
|
||||
def get_locations(self, all: bool = False, limit: int = 1000, **kwargs) -> list[Location]:
|
||||
url = "locations/"
|
||||
filters = self.handle_kwargs(kwargs)
|
||||
response = self.get_all_request(url, limit, filters) if all else self.get_request(url, filters)
|
||||
return ComicVineResponse(**response).handle_result(Location)
|
||||
|
||||
def get_movie(self, id: int) -> Movie:
|
||||
url = f"movie/4025-{id}"
|
||||
return ComicVineResponse(**self.get_request(url)).handle_result(Movie)
|
||||
|
||||
def get_movies(self, all: bool = False, limit: int = 1000, **kwargs) -> list[Movie]:
|
||||
url = "movies/"
|
||||
filters = self.handle_kwargs(kwargs)
|
||||
response = self.get_all_request(url, limit, filters) if all else self.get_request(url, filters)
|
||||
return ComicVineResponse(**response).handle_result(Movie)
|
||||
|
||||
def get_object(self, id: int) -> Object:
|
||||
url = f"object/4055-{id}"
|
||||
return ComicVineResponse(**self.get_request(url)).handle_result(Object)
|
||||
|
||||
def get_objects(self, all: bool = False, limit: int = 1000, **kwargs) -> list[Object]:
|
||||
url = "objects/"
|
||||
filters = self.handle_kwargs(kwargs)
|
||||
response = self.get_all_request(url, limit, filters) if all else self.get_request(url, filters)
|
||||
return ComicVineResponse(**response).handle_result(Object)
|
||||
|
||||
def get_person(self, id: int) -> Person:
|
||||
url = f"person/4040-{id}"
|
||||
return ComicVineResponse(**self.get_request(url)).handle_result(Person)
|
||||
|
||||
def get_people(self, all: bool = False, limit: int = 1000, **kwargs) -> list[People]:
|
||||
url = "people/"
|
||||
filters = self.handle_kwargs(kwargs)
|
||||
response = self.get_all_request(url, limit, filters) if all else self.get_request(url, filters)
|
||||
return ComicVineResponse(**response).handle_result(People)
|
||||
|
||||
def get_power(self, id: int) -> Power:
|
||||
url = f"power/4035-{id}"
|
||||
return ComicVineResponse(**self.get_request(url)).handle_result(Power)
|
||||
|
||||
def get_powers(self, all: bool = False, limit: int = 1000, **kwargs) -> list[Power]:
|
||||
url = "powers/"
|
||||
filters = self.handle_kwargs(kwargs)
|
||||
response = self.get_all_request(url, limit, filters) if all else self.get_request(url, filters)
|
||||
return ComicVineResponse(**response).handle_result(Power)
|
||||
|
||||
def get_promo(self, id: int) -> Promo:
|
||||
url = f"promo/1700-{id}"
|
||||
return ComicVineResponse(**self.get_request(url)).handle_result(Promo)
|
||||
|
||||
def get_promos(self, all: bool = False, limit: int = 1000, **kwargs) -> list[Promo]:
|
||||
url = "promos/"
|
||||
filters = self.handle_kwargs(kwargs)
|
||||
response = self.get_all_request(url, limit, filters) if all else self.get_request(url, filters)
|
||||
return ComicVineResponse(**response).handle_result(Promo)
|
||||
|
||||
def get_publisher(self, id: int) -> Publisher:
|
||||
url = f"publisher/4010-{id}"
|
||||
return ComicVineResponse(**self.get_request(url)).handle_result(Publisher)
|
||||
|
||||
def get_publishers(self, all: bool = False, limit: int = 1000, **kwargs) -> list[Publisher]:
|
||||
url = "publishers/"
|
||||
filters = self.handle_kwargs(kwargs)
|
||||
response = self.get_all_request(url, limit, filters) if all else self.get_request(url, filters)
|
||||
return ComicVineResponse(**response).handle_result(Publisher)
|
||||
|
||||
def get_series(self, all: bool = False, limit: int = 1000, **kwargs) -> list[Series]:
|
||||
url = "series/"
|
||||
filters = self.handle_kwargs(kwargs)
|
||||
response = self.get_all_request(url, limit, filters) if all else self.get_request(url, filters)
|
||||
return ComicVineResponse(**response).handle_result(Series)
|
||||
|
||||
def get_series_list(self, all: bool = False, limit: int = 1000, **kwargs) -> list[Series]:
|
||||
url = "series_list/"
|
||||
filters = self.handle_kwargs(kwargs)
|
||||
response = self.get_all_request(url, limit, filters) if all else self.get_request(url, filters)
|
||||
return ComicVineResponse(**response).handle_result(Series)
|
||||
|
||||
def get_story_arc(self, id: int) -> StoryArc:
|
||||
url = f"story_arc/4045-{id}"
|
||||
return ComicVineResponse(**self.get_request(url)).handle_result(StoryArc)
|
||||
|
||||
def get_story_arcs(self, all: bool = False, limit: int = 1000, **kwargs) -> list[StoryArcs]:
|
||||
url = "story_arcs/"
|
||||
filters = self.handle_kwargs(kwargs)
|
||||
response = self.get_all_request(url, limit, filters) if all else self.get_request(url, filters)
|
||||
return ComicVineResponse(**response).handle_result(StoryArcs)
|
||||
|
||||
def get_team(self, id: int) -> Team:
|
||||
url = f"team/4060-{id}"
|
||||
return ComicVineResponse(**self.get_request(url)).handle_result(Team)
|
||||
|
||||
def get_teams(self, all: bool = False, limit: int = 1000, **kwargs) -> list[Teams]:
|
||||
url = "teams/"
|
||||
filters = self.handle_kwargs(kwargs)
|
||||
response = self.get_all_request(url, limit, filters) if all else self.get_request(url, filters)
|
||||
return ComicVineResponse(**response).handle_result(Teams)
|
||||
|
||||
# TODOs kept as in original
|
||||
def get_types(self, **kwargs) -> list[Type]:
|
||||
pass
|
||||
|
||||
def get_video(self, id: int) -> Video:
|
||||
pass
|
||||
|
||||
def get_videos(self, **kwargs) -> list[Video]:
|
||||
pass
|
||||
|
||||
def get_video_type(self, id: int) -> VideoType:
|
||||
pass
|
||||
|
||||
def get_video_types(self, **kwargs) -> list[VideoType]:
|
||||
pass
|
||||
|
||||
def get_video_category(self, id: int) -> VideoCategory:
|
||||
pass
|
||||
|
||||
def get_video_categories(self, **kwargs) -> list[VideoCategory]:
|
||||
pass
|
||||
|
||||
def get_volumes(
|
||||
self,
|
||||
all: bool = True,
|
||||
limit: int = 1000,
|
||||
sort: str = "asc",
|
||||
offset: int = 0,
|
||||
**kwargs: Any,
|
||||
) -> list[Volumes]:
|
||||
url = "volumes/"
|
||||
filters = self.handle_kwargs(kwargs)
|
||||
filters["sort"] = sort
|
||||
filters["offset"] = str(offset)
|
||||
log.debug(f"Getting volumes with filters: {filters}")
|
||||
response = self.get_all_request(url, limit, filters) if all else self.get_request(url, filters)
|
||||
return ComicVineResponse(**response).handle_result(Volumes)
|
||||
|
||||
def get_volume(self, id: int, **kwargs) -> Volume:
|
||||
url = f"volume/4050-{id}"
|
||||
filters = self.handle_kwargs(kwargs)
|
||||
log.debug(f"Getting volume with filters: {filters}")
|
||||
response = self.get_request(url, filters)
|
||||
return ComicVineResponse(**response).handle_result(Volume)
|
||||
83
src/comicvineapi/cache.py
Normal file
83
src/comicvineapi/cache.py
Normal file
@@ -0,0 +1,83 @@
|
||||
# create a cache that stores the results in a database
|
||||
# use the dataclasses to create the tables
|
||||
|
||||
from comicvineapi.schemas.api_classes import *
|
||||
import sqlite3
|
||||
from dataclasses import dataclass, fields
|
||||
from loguru import logger
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
from typing import Any, Optional
|
||||
import json
|
||||
import os
|
||||
|
||||
from datetime import datetime, timedelta, timezone
|
||||
|
||||
|
||||
log = logger
|
||||
log.remove()
|
||||
log.add(
|
||||
f"logs/database_{datetime.now().strftime('%Y-%m-%d')}.log",
|
||||
rotation="1 day",
|
||||
compression="zip",
|
||||
)
|
||||
log.add("logs/database.log", compression="zip", rotation="50 MB")
|
||||
def cache_root():
|
||||
cache_location = os.getenv("XDG_CACHE_HOME", default=str(Path.home() / ".cache"))
|
||||
folder = Path(cache_location).resolve() / "comicvineAPI"
|
||||
folder.mkdir(parents=True, exist_ok=True)
|
||||
return folder
|
||||
|
||||
|
||||
class Cache:
|
||||
"""A simple cache for storing and retrieving API responses using SQLite."""
|
||||
def __init__(
|
||||
self, cache_location: Optional[Path] = None, store_duration: Optional[int] = 10
|
||||
):
|
||||
self.keep_for = store_duration
|
||||
self.connection = sqlite3.connect(cache_location or cache_root() / "cache.db")
|
||||
|
||||
self.cursor = self.connection.cursor()
|
||||
self.create_table()
|
||||
self.remove_expired()
|
||||
|
||||
def create_table(self):
|
||||
self.connection.execute(
|
||||
"CREATE TABLE IF NOT EXISTS requests (id INTEGER PRIMARY KEY AUTOINCREMENT, query, result, timestamp)"
|
||||
)
|
||||
self.connection.commit()
|
||||
self.remove_expired()
|
||||
|
||||
def insert_response(self, query: str, result: Any):
|
||||
self.connection.execute(
|
||||
"INSERT INTO requests (query, result, timestamp) VALUES (?, ?, ?)",
|
||||
(query, json.dumps(result), datetime.now(timezone.utc)),
|
||||
)
|
||||
self.connection.commit()
|
||||
logger.info(f"Inserted {query} into cache")
|
||||
|
||||
def get_response(self, query: str):
|
||||
log.debug(f"Getting {query} from cache")
|
||||
if self.keep_for:
|
||||
expity_date = datetime.now(timezone.utc) - timedelta(days=self.keep_for)
|
||||
conn = self.connection.execute(
|
||||
"SELECT result FROM requests WHERE query = ? AND timestamp > ?",
|
||||
(query, expity_date),
|
||||
)
|
||||
else:
|
||||
conn = self.connection.execute(
|
||||
"SELECT result FROM requests WHERE query = ?", (query,)
|
||||
)
|
||||
if result := conn.fetchone():
|
||||
log.success(f"Got {query} from cache")
|
||||
return json.loads(result[0])
|
||||
return None
|
||||
|
||||
def remove_expired(self):
|
||||
if not self.keep_for:
|
||||
return
|
||||
expiry_date = datetime.now(timezone.utc) - timedelta(days=self.keep_for)
|
||||
self.connection.execute(
|
||||
"DELETE FROM requests WHERE timestamp < ?", (expiry_date,)
|
||||
)
|
||||
self.connection.commit()
|
||||
64
src/comicvineapi/response.py
Normal file
64
src/comicvineapi/response.py
Normal file
@@ -0,0 +1,64 @@
|
||||
from dataclasses import dataclass, field
|
||||
from loguru import logger
|
||||
from datetime import datetime
|
||||
|
||||
log = logger
|
||||
log.remove()
|
||||
log.add(
|
||||
f"logs/{datetime.now().strftime('%Y-%m-%d')}.log",
|
||||
rotation="1 day",
|
||||
compression="zip",
|
||||
)
|
||||
log.add("logs/api.log", compression="zip", rotation="50 MB")
|
||||
|
||||
|
||||
@dataclass
|
||||
class ComicVineResponse:
|
||||
status_code: int = 0
|
||||
error: str | None = None
|
||||
number_of_total_results: str | None = None
|
||||
number_of_page_results: str | None = None
|
||||
limit: str | None = None
|
||||
offset: str | None = None
|
||||
results: list = field(default_factory=list)
|
||||
version: float | None = None
|
||||
|
||||
def __post_init__(self):
|
||||
self.status = self.translate_status_code()
|
||||
log.debug(
|
||||
f"Response status: {self.status}, got {self.number_of_page_results} pages of {self.number_of_total_results} total results"
|
||||
)
|
||||
|
||||
def translate_status_code(self):
|
||||
match self.status_code:
|
||||
case 1:
|
||||
return "OK"
|
||||
case 100:
|
||||
raise Exception("Invalid API Key")
|
||||
case 101:
|
||||
raise Exception("Object Not Found")
|
||||
case 102:
|
||||
raise Exception("Error in URL Format")
|
||||
case 103:
|
||||
raise Exception("'jsonp' format requires a 'json_callback' argument")
|
||||
case 104:
|
||||
raise Exception("Filter Error")
|
||||
case 105:
|
||||
raise Exception("Subscriber only video is for subscribers only")
|
||||
|
||||
def handle_result(self, resultType):
|
||||
log.debug(f"Handling result, result is: {type(self.results)}")
|
||||
if self.number_of_page_results == 0:
|
||||
return None
|
||||
if self.number_of_total_results == 1:
|
||||
return resultType(**self.results)
|
||||
results = [None] * len(self.results)
|
||||
for i, result in enumerate(self.results):
|
||||
if isinstance(result, list):
|
||||
continue
|
||||
else:
|
||||
results[i] = resultType(**result)
|
||||
# remove any None values
|
||||
results = [x for x in results if x is not None]
|
||||
log.debug(f"Returning {len(results)} result(s)")
|
||||
return results
|
||||
2
src/comicvineapi/schemas/__init__.py
Normal file
2
src/comicvineapi/schemas/__init__.py
Normal file
@@ -0,0 +1,2 @@
|
||||
from .response import ComicVineResponse
|
||||
from .api_classes import *
|
||||
970
src/comicvineapi/schemas/api_classes.py
Normal file
970
src/comicvineapi/schemas/api_classes.py
Normal file
@@ -0,0 +1,970 @@
|
||||
from dataclasses import dataclass
|
||||
from datetime import datetime
|
||||
from loguru import logger
|
||||
from datetime import datetime
|
||||
from typing import Any
|
||||
|
||||
log = logger
|
||||
log.remove()
|
||||
log.add(
|
||||
f"logs/{datetime.now().strftime('%Y-%m-%d')}.log",
|
||||
rotation="1 day",
|
||||
compression="zip",
|
||||
)
|
||||
log.add("logs/api.log", compression="zip", rotation="50 MB")
|
||||
|
||||
|
||||
@dataclass
|
||||
class Characters:
|
||||
aliases: str | list | None = None
|
||||
api_detail_url: str | None = None
|
||||
birth: str | None = None
|
||||
count_of_issue_appearances: int | None = None
|
||||
date_added: str | None = None
|
||||
date_last_updated: str | None = None
|
||||
deck: str | None = None
|
||||
description: str | None = None
|
||||
first_appeared_in_issue: dict | None = None
|
||||
gender: int | None = None
|
||||
id: int | None = None
|
||||
name: str | None = None
|
||||
image: dict | None = None
|
||||
origin: dict | None = None
|
||||
publisher: dict | None = None
|
||||
real_name: str | None = None
|
||||
site_detail_url: str | None = None
|
||||
|
||||
def __post_init__(self):
|
||||
if isinstance(self.aliases, str):
|
||||
self.aliases = [alias for alias in self.aliases.split("\n")]
|
||||
if isinstance(self.origin, dict):
|
||||
self.origin = Origin(**self.origin)
|
||||
if isinstance(self.publisher, dict):
|
||||
self.publisher = Publisher(**self.publisher)
|
||||
if isinstance(self.image, dict):
|
||||
self.image = Image(**self.image)
|
||||
if isinstance(self.date_added, str):
|
||||
self.date_added = datetime.strptime(self.date_added, "%Y-%m-%d %H:%M:%S")
|
||||
if isinstance(self.date_last_updated, str):
|
||||
self.date_last_updated = datetime.strptime(
|
||||
self.date_last_updated, "%Y-%m-%d %H:%M:%S"
|
||||
)
|
||||
|
||||
# def __table__(self):
|
||||
# return self.__dict__
|
||||
|
||||
|
||||
@dataclass
|
||||
class Character(Characters):
|
||||
character_enemies: list | None = None
|
||||
character_friends: list | None = None
|
||||
count_of_issue_appearances: int | None = None
|
||||
date_added: str | None = None
|
||||
date_last_updated: str | None = None
|
||||
deck: str | None = None
|
||||
description: str | None = None
|
||||
first_appeared_in_issue: dict | None = None
|
||||
creators: list | None = None
|
||||
issue_credits: list | None = None
|
||||
issues_died_in: list | None = None
|
||||
movies: list | None = None
|
||||
powers: list | None = None
|
||||
story_arc_credits: list | None = None
|
||||
team_enemies: list | None = None
|
||||
team_friends: list | None = None
|
||||
teams: list | None = None
|
||||
volume_credits: list | None = None
|
||||
count:int|None=None
|
||||
|
||||
def __post_init__(self):
|
||||
if isinstance(self.character_enemies, list):
|
||||
self.character_enemies = [
|
||||
Character(**character) for character in self.character_enemies
|
||||
]
|
||||
if isinstance(self.character_friends, list):
|
||||
self.character_friends = [
|
||||
Character(**character) for character in self.character_friends
|
||||
]
|
||||
if isinstance(self.creators, list):
|
||||
self.creators = [Person(**creator) for creator in self.creators]
|
||||
if isinstance(self.issue_credits, list):
|
||||
self.issue_credits = [Issue(**issue) for issue in self.issue_credits]
|
||||
if isinstance(self.issues_died_in, list):
|
||||
self.issues_died_in = [Issue(**issue) for issue in self.issues_died_in]
|
||||
if isinstance(self.movies, list):
|
||||
self.movies = [Movie(**movie) for movie in self.movies]
|
||||
if isinstance(self.powers, list):
|
||||
self.powers = [Power(**power) for power in self.powers]
|
||||
if isinstance(self.story_arc_credits, list):
|
||||
self.story_arc_credits = [StoryArc(**arc) for arc in self.story_arc_credits]
|
||||
if isinstance(self.team_enemies, list):
|
||||
self.team_enemies = [Team(**team) for team in self.team_enemies]
|
||||
if isinstance(self.team_friends, list):
|
||||
self.team_friends = [Team(**team) for team in self.team_friends]
|
||||
if isinstance(self.volume_credits, list):
|
||||
self.volume_credits = [Volume(**volume) for volume in self.volume_credits]
|
||||
if isinstance(self.teams, list):
|
||||
self.teams = [Team(**team) for team in self.teams]
|
||||
|
||||
|
||||
@dataclass
|
||||
class Chat:
|
||||
api_detail_url: str | None = None
|
||||
channel_name: str | None = None
|
||||
deck: str | None = None
|
||||
image: str | None = None
|
||||
password: str | None = None
|
||||
site_detail_url: str | None = None
|
||||
title: str | None = None
|
||||
|
||||
|
||||
|
||||
@dataclass
|
||||
class Concepts:
|
||||
aliases: str | list | None = None
|
||||
api_detail_url: str | None = None
|
||||
count_of_issue_appearances: int | None = None
|
||||
date_added: str | None = None
|
||||
date_last_updated: str | None = None
|
||||
deck: str | None = None
|
||||
description: str | None = None
|
||||
first_appeared_in_issue: str | None = None
|
||||
id: int | None = None
|
||||
image: str | None = None
|
||||
name: str | None = None
|
||||
issue_credits: list | None = None
|
||||
site_detail_url: str | None = None
|
||||
start_year: int | None = None
|
||||
volume_credits: list | None = None
|
||||
|
||||
def __post_init__(self):
|
||||
if isinstance(self.aliases, str):
|
||||
self.aliases = [alias for alias in self.aliases.split("\n")]
|
||||
if isinstance(self.date_added, str):
|
||||
self.date_added = datetime.strptime(self.date_added, "%Y-%m-%d %H:%M:%S")
|
||||
if isinstance(self.date_last_updated, str):
|
||||
self.date_last_updated = datetime.strptime(
|
||||
self.date_last_updated, "%Y-%m-%d %H:%M:%S"
|
||||
)
|
||||
if isinstance(self.volume_credits, list):
|
||||
self.volume_credits = [Volume(**volume) for volume in self.volume_credits]
|
||||
if isinstance(self.issue_credits, list):
|
||||
self.issue_credits = [Issue(**issue) for issue in self.issue_credits]
|
||||
if isinstance(self.image, str):
|
||||
self.image = Image(original_url=self.image)
|
||||
|
||||
@dataclass
|
||||
class Concept(Concepts):
|
||||
count: int | None = None
|
||||
|
||||
|
||||
|
||||
@dataclass
|
||||
class Episodes:
|
||||
aliases: str | list | None = None
|
||||
api_detail_url: str | None = None
|
||||
date_added: str | None = None
|
||||
date_last_updated: str | None = None
|
||||
deck: str | None = None
|
||||
description: str | None = None
|
||||
episode_number: int | None = None
|
||||
id: int | None = None
|
||||
image: str | None = None
|
||||
name: str | None = None
|
||||
site_detail_url: str | None = None
|
||||
air_date: str | None = None
|
||||
has_staff_review: dict | None = None
|
||||
series: dict | None = None
|
||||
|
||||
def __post_init__(self):
|
||||
if isinstance(self.aliases, str):
|
||||
self.aliases = [alias for alias in self.aliases.split("\n")]
|
||||
if isinstance(self.air_date, str):
|
||||
self.air_date = datetime.strptime(self.air_date, "%Y-%m-%d")
|
||||
if isinstance(self.date_added, str):
|
||||
self.date_added = datetime.strptime(self.date_added, "%Y-%m-%d %H:%M:%S")
|
||||
if isinstance(self.date_last_updated, str):
|
||||
self.date_last_updated = datetime.strptime(
|
||||
self.date_last_updated, "%Y-%m-%d %H:%M:%S"
|
||||
)
|
||||
if isinstance(self.volume, dict):
|
||||
self.volume = Volume(**self.volume)
|
||||
if isinstance(self.volume_credits, list):
|
||||
self.volume_credits = [Volume(**volume) for volume in self.volume_credits]
|
||||
if isinstance(self.has_staff_review, dict):
|
||||
self.has_staff_review = StaffReview(**self.has_staff_review)
|
||||
if isinstance(self.series, dict):
|
||||
self.series = Series(**self.series)
|
||||
|
||||
|
||||
@dataclass
|
||||
class Episode(Episodes):
|
||||
character_credits: list | None = None
|
||||
character_died_in: list | None = None
|
||||
concept_credits: list | None = None
|
||||
first_appearance_characters: list | None = None
|
||||
first_appearance_concepts: Any | None = None
|
||||
first_appearance_locations: list | None = None
|
||||
first_appearance_objects: list | None = None
|
||||
first_appearance_storyarcs: list | None = None
|
||||
first_appearance_teams: list | None = None
|
||||
location_credits: list | None = None
|
||||
object_credits: list | None = None
|
||||
story_arc_credits: list | None = None
|
||||
team_credits: list | None = None
|
||||
|
||||
def __post_init__(self):
|
||||
if isinstance(self.character_credits, list):
|
||||
self.character_credits = [
|
||||
Character(**character) for character in self.character_credits
|
||||
]
|
||||
if isinstance(self.character_died_in, list):
|
||||
self.character_died_in = [
|
||||
Character(**character) for character in self.character_died_in
|
||||
]
|
||||
if isinstance(self.concept_credits, list):
|
||||
self.concept_credits = [
|
||||
Concept(**concept) for concept in self.concept_credits
|
||||
]
|
||||
if isinstance(self.first_appearance_characters, list):
|
||||
self.first_appearance_characters = [
|
||||
Character(**character) for character in self.first_appearance_characters
|
||||
]
|
||||
if isinstance(self.first_appearance_concepts, list):
|
||||
self.first_appearance_concepts = [
|
||||
Concept(**concept) for concept in self.first_appearance_concepts
|
||||
]
|
||||
if isinstance(self.first_appearance_locations, list):
|
||||
self.first_appearance_locations = [
|
||||
Location(**location) for location in self.first_appearance_locations
|
||||
]
|
||||
if isinstance(self.first_appearance_objects, list):
|
||||
self.first_appearance_objects = [
|
||||
Object(**object) for object in self.first_appearance_objects
|
||||
]
|
||||
if isinstance(self.first_appearance_storyarcs, list):
|
||||
self.first_appearance_storyarcs = [
|
||||
StoryArc(**arc) for arc in self.first_appearance_storyarcs
|
||||
]
|
||||
if isinstance(self.first_appearance_teams, list):
|
||||
self.first_appearance_teams = [
|
||||
Team(**team) for team in self.first_appearance_teams
|
||||
]
|
||||
if isinstance(self.location_credits, list):
|
||||
self.location_credits = [
|
||||
Location(**location) for location in self.location_credits
|
||||
]
|
||||
if isinstance(self.object_credits, list):
|
||||
self.object_credits = [Object(**object) for object in self.object_credits]
|
||||
if isinstance(self.story_arc_credits, list):
|
||||
self.story_arc_credits = [StoryArc(**arc) for arc in self.story_arc_credits]
|
||||
if isinstance(self.team_credits, list):
|
||||
self.team_credits = [Team(**team) for team in self.team_credits]
|
||||
|
||||
|
||||
@dataclass
|
||||
class Image:
|
||||
icon_url: str | None = None
|
||||
medium_url: str | None = None
|
||||
screen_url: str | None = None
|
||||
screen_large_url: str | None = None
|
||||
small_url: str | None = None
|
||||
super_url: str | None = None
|
||||
thumb_url: str | None = None
|
||||
tiny_url: str | None = None
|
||||
image_tags: str | None = None
|
||||
original_url: str | None = None
|
||||
|
||||
def __str__(self):
|
||||
return self.original_url
|
||||
|
||||
def __post__init__(self):
|
||||
self.image_tags = self.image_tags.split(",") if self.image_tags else None
|
||||
|
||||
|
||||
@dataclass
|
||||
class Issues:
|
||||
aliases: str | list | None = None
|
||||
api_detail_url: str | None = None
|
||||
cover_date: str | None = None
|
||||
date_added: str | None = None
|
||||
date_last_updated: str | None = None
|
||||
deck: str | None = None
|
||||
description: str | None = None
|
||||
has_staff_review: bool | None = None
|
||||
id: int | None = None
|
||||
image: dict | None = None
|
||||
issue_number: str | None = None
|
||||
name: str | None = None
|
||||
site_detail_url: str | None = None
|
||||
store_date: str | None = None
|
||||
volume: dict | None = None
|
||||
associated_images: list | None = None
|
||||
|
||||
def __post_init__(self):
|
||||
if isinstance(self.aliases, str):
|
||||
self.aliases = [alias for alias in self.aliases.split("\n")]
|
||||
if isinstance(self.date_added, str):
|
||||
self.date_added = datetime.strptime(self.date_added, "%Y-%m-%d %H:%M:%S")
|
||||
if isinstance(self.date_last_updated, str):
|
||||
self.date_last_updated = datetime.strptime(
|
||||
self.date_last_updated, "%Y-%m-%d %H:%M:%S"
|
||||
)
|
||||
if isinstance(self.image, dict):
|
||||
self.image = Image(**self.image)
|
||||
if isinstance(self.volume, dict):
|
||||
self.volume = Volume(**self.volume)
|
||||
if isinstance(self.associated_images, list):
|
||||
self.associated_images = [
|
||||
Image(**image) for image in self.associated_images
|
||||
]
|
||||
|
||||
|
||||
@dataclass
|
||||
class Issue(Issues):
|
||||
character_credits: list | None = None
|
||||
character_died_in: list | None = None
|
||||
concept_credits: list | None = None
|
||||
first_appearance_characters: list | None = None
|
||||
first_appearance_concepts: list | None = None
|
||||
first_appearance_locations: list | None = None
|
||||
first_appearance_objects: list | None = None
|
||||
first_appearance_storyarcs: list | None = None
|
||||
first_appearance_teams: list | None = None
|
||||
location_credits: list | None = None
|
||||
object_credits: list | None = None
|
||||
person_credits: list | None = None
|
||||
story_arc_credits: list | None = None
|
||||
team_credits: list | None = None
|
||||
team_disbanded_in: list | None = None
|
||||
|
||||
def __post_init__(self):
|
||||
if isinstance(self.character_credits, list):
|
||||
self.character_credits = [
|
||||
Character(**character) for character in self.character_credits
|
||||
]
|
||||
if isinstance(self.character_died_in, list):
|
||||
self.character_died_in = [
|
||||
Character(**character) for character in self.character_died_in
|
||||
]
|
||||
if isinstance(self.concept_credits, list):
|
||||
self.concept_credits = [
|
||||
Concept(**concept) for concept in self.concept_credits
|
||||
]
|
||||
if isinstance(self.first_appearance_characters, list):
|
||||
self.first_appearance_characters = [
|
||||
Character(**character) for character in self.first_appearance_characters
|
||||
]
|
||||
if isinstance(self.first_appearance_concepts, list):
|
||||
self.first_appearance_concepts = [
|
||||
Concept(**concept) for concept in self.first_appearance_concepts
|
||||
]
|
||||
if isinstance(self.first_appearance_locations, list):
|
||||
self.first_appearance_locations = [
|
||||
Location(**location) for location in self.first_appearance_locations
|
||||
]
|
||||
if isinstance(self.first_appearance_objects, list):
|
||||
self.first_appearance_objects = [
|
||||
Object(**object) for object in self.first_appearance_objects
|
||||
]
|
||||
if isinstance(self.first_appearance_storyarcs, list):
|
||||
self.first_appearance_storyarcs = [
|
||||
StoryArc(**arc) for arc in self.first_appearance_storyarcs
|
||||
]
|
||||
if isinstance(self.first_appearance_teams, list):
|
||||
self.first_appearance_teams = [
|
||||
Team(**team) for team in self.first_appearance_teams
|
||||
]
|
||||
if isinstance(self.location_credits, list):
|
||||
self.location_credits = [
|
||||
Location(**location) for location in self.location_credits
|
||||
]
|
||||
if isinstance(self.object_credits, list):
|
||||
self.object_credits = [Object(**object) for object in self.object_credits]
|
||||
if isinstance(self.person_credits, list):
|
||||
self.person_credits = [Person(**person) for person in self.person_credits]
|
||||
if isinstance(self.story_arc_credits, list):
|
||||
self.story_arc_credits = [StoryArc(**arc) for arc in self.story_arc_credits]
|
||||
if isinstance(self.team_credits, list):
|
||||
self.team_credits = [Team(**team) for team in self.team_credits]
|
||||
if isinstance(self.team_disbanded_in, list):
|
||||
self.team_disbanded_in = [Team(**team) for team in self.team_disbanded_in]
|
||||
|
||||
|
||||
@dataclass
|
||||
class Location:
|
||||
aliases: str | list | None = None
|
||||
api_detail_url: str | None = None
|
||||
count_of_issue_appearances: int | None = None
|
||||
date_added: str | None | datetime = None
|
||||
date_last_updated: str | None | datetime = None
|
||||
deck: str | None = None
|
||||
description: str | None = None
|
||||
first_appeared_in_issue: str | None = None
|
||||
id: int | None = None
|
||||
image: Image | dict | None = None
|
||||
issue_credits: list | None = None
|
||||
movies: str | None = None
|
||||
name: str | None = None
|
||||
site_detail_url: str | None = None
|
||||
start_year: int | None = None
|
||||
story_arc_credits: list | None = None
|
||||
volume_credits: list | None = None
|
||||
count:int|None=None
|
||||
def __post_init__(self):
|
||||
if isinstance(self.aliases, str):
|
||||
self.aliases = [alias for alias in self.aliases.split("\n")]
|
||||
if isinstance(self.image, dict):
|
||||
self.image = Image(**self.image)
|
||||
if isinstance(self.date_added, str):
|
||||
self.date_added = datetime.strptime(self.date_added, "%Y-%m-%d %H:%M:%S")
|
||||
if isinstance(self.date_last_updated, str):
|
||||
self.date_last_updated = datetime.strptime(
|
||||
self.date_last_updated, "%Y-%m-%d %H:%M:%S"
|
||||
)
|
||||
|
||||
|
||||
@dataclass
|
||||
class Movie:
|
||||
api_detail_url: str | None = None
|
||||
box_office_revenue: str | None = None
|
||||
budget: str | None = None
|
||||
characters: str | None | list[Character] = None
|
||||
concepts: str | None | list["Concept"] = None
|
||||
date_added: str | None = None
|
||||
date_last_updated: str | None = None
|
||||
deck: str | None = None
|
||||
description: str | None = None
|
||||
distributors: str | None = None
|
||||
has_staff_review: bool | None = None
|
||||
id: int | None = None
|
||||
image: dict | None | Image = None
|
||||
locations: list | None | list[Location] = None
|
||||
name: str | None = None
|
||||
producers: list | None = None
|
||||
rating: str | None = None
|
||||
release_date: str | None = None
|
||||
runtime: str | None = None
|
||||
site_detail_url: str | None = None
|
||||
studios: list | None | list["Studio"] = None
|
||||
teams: str | None = None
|
||||
things: str | None = None
|
||||
total_revenue: str | None = None
|
||||
writers: list | None | list["Writer"] = None
|
||||
|
||||
def __post_init__(self):
|
||||
if isinstance(self.image, dict):
|
||||
self.image = Image(**self.image)
|
||||
if isinstance(self.date_added, str):
|
||||
self.date_added = datetime.strptime(self.date_added, "%Y-%m-%d %H:%M:%S")
|
||||
if isinstance(self.date_last_updated, str):
|
||||
self.date_last_updated = datetime.strptime(
|
||||
self.date_last_updated, "%Y-%m-%d %H:%M:%S"
|
||||
)
|
||||
if isinstance(self.release_date, str):
|
||||
self.release_date = datetime.strptime(
|
||||
self.release_date, "%Y-%m-%d %H:%M:%S"
|
||||
)
|
||||
if isinstance(self.characters, str):
|
||||
self.characters = [Character(**character) for character in self.characters]
|
||||
if isinstance(self.concepts, str):
|
||||
self.concepts = [Concept(**concept) for concept in self.concepts]
|
||||
if isinstance(self.locations, str):
|
||||
self.locations = [Location(**location) for location in self.locations]
|
||||
if isinstance(self.teams, str):
|
||||
self.teams = [Team(**team) for team in self.teams]
|
||||
if isinstance(self.producers, list):
|
||||
self.producers = [Producer(**producer) for producer in self.producers]
|
||||
if isinstance(self.studios, list):
|
||||
self.studios = [Studio(**studio) for studio in self.studios]
|
||||
if isinstance(self.writers, list):
|
||||
self.writers = [Writer(**writer) for writer in self.writers]
|
||||
|
||||
|
||||
@dataclass
|
||||
class Object:
|
||||
aliases: str | None | list = None
|
||||
api_detail_url: str | None = None
|
||||
count_of_issue_appearances: int | None = None
|
||||
date_added: str | None | datetime = None
|
||||
date_last_updated: str | None | datetime = None
|
||||
deck: str | None = None
|
||||
description: str | None = None
|
||||
first_appeared_in_issue: str | None = None
|
||||
id: int | None = None
|
||||
image: str | None = None
|
||||
issue_credits: list | None = None
|
||||
movies: str | None | list["Movie"] = None
|
||||
name: str | None = None
|
||||
site_detail_url: str | None = None
|
||||
start_year: int | None = None
|
||||
story_arc_credits: list | None = None
|
||||
volume_credits: list | None = None
|
||||
count:int|None=None
|
||||
def __post_init__(self):
|
||||
if isinstance(self.aliases, str):
|
||||
self.aliases = [alias for alias in self.aliases.split("\n")]
|
||||
if isinstance(self.date_added, str):
|
||||
self.date_added = datetime.strptime(self.date_added, "%Y-%m-%d %H:%M:%S")
|
||||
if isinstance(self.date_last_updated, str):
|
||||
self.date_last_updated = datetime.strptime(
|
||||
self.date_last_updated, "%Y-%m-%d %H:%M:%S"
|
||||
)
|
||||
if isinstance(self.movies, list):
|
||||
self.movies = [Movie(**movie) for movie in self.movies]
|
||||
|
||||
|
||||
@dataclass
|
||||
class Origin:
|
||||
api_detail_url: str | None = None
|
||||
character_set: str | None = None
|
||||
id: int | None = None
|
||||
name: str | None = None
|
||||
profiles: str | None = None
|
||||
site_detail_url: str | None = None
|
||||
|
||||
def __str__(self):
|
||||
return self.name
|
||||
|
||||
|
||||
@dataclass
|
||||
class People:
|
||||
aliases: str | list | None = None
|
||||
api_detail_url: str | None = None
|
||||
birth: str | None = None
|
||||
count_of_isssue_appearances: int | None = None
|
||||
country: str | None = None
|
||||
created_characters: str | None = None
|
||||
date_added: str | None | datetime = None
|
||||
date_last_updated: str | None | datetime = None
|
||||
death: str | None = None
|
||||
deck: str | None = None
|
||||
description: str | None = None
|
||||
email: str | None = None
|
||||
gender: int | str | None = None
|
||||
hometown: str | None = None
|
||||
id: int | None = None
|
||||
image: str | Image | None = None
|
||||
issue_credits: list | None = None
|
||||
name: str | None = None
|
||||
site_detail_url: str | None = None
|
||||
story_arc_credits: list | None = None
|
||||
volume_credits: list | None = None
|
||||
website: str | None = None
|
||||
role: str | None = None
|
||||
|
||||
def __post_init__(self):
|
||||
if isinstance(self.aliases, str):
|
||||
self.aliases = [alias for alias in self.aliases.split("\n")]
|
||||
if isinstance(self.image, dict):
|
||||
self.image = Image(**self.image)
|
||||
if isinstance(self.date_added, str):
|
||||
self.date_added = datetime.strptime(self.date_added, "%Y-%m-%d %H:%M:%S")
|
||||
if isinstance(self.date_last_updated, str):
|
||||
self.date_last_updated = datetime.strptime(
|
||||
self.date_last_updated, "%Y-%m-%d %H:%M:%S"
|
||||
)
|
||||
if isinstance(self.gender, int):
|
||||
match self.gender:
|
||||
case 1:
|
||||
self.gender = "male"
|
||||
case 2:
|
||||
self.gender = "female"
|
||||
case _:
|
||||
self.gender = "other"
|
||||
if isinstance(self.issue_credits, list):
|
||||
self.issue_credits = [Issue(**issue) for issue in self.issue_credits]
|
||||
if isinstance(self.story_arc_credits, list):
|
||||
self.story_arc_credits = [StoryArc(**arc) for arc in self.story_arc_credits]
|
||||
if isinstance(self.volume_credits, list):
|
||||
self.volume_credits = [Volume(**volume) for volume in self.volume_credits]
|
||||
if isinstance(self.story_arc_credits, list):
|
||||
self.story_arc_credits = [StoryArc(**arc) for arc in self.story_arc_credits]
|
||||
|
||||
|
||||
@dataclass
|
||||
class Person(People):
|
||||
issues: list | None = None
|
||||
count:int|None=None
|
||||
|
||||
def __post_init__(self):
|
||||
if isinstance(self.issues, list):
|
||||
self.issues = [Issue(**issue) for issue in self.issues]
|
||||
|
||||
|
||||
@dataclass
|
||||
class Power:
|
||||
aliases: str | list | None = None
|
||||
api_detail_url: str | None = None
|
||||
characters: str | list[Character] | None = None
|
||||
date_added: datetime | str | None = None
|
||||
date_last_updated: datetime | str | None = None
|
||||
description: str | None = None
|
||||
id: int | None = None
|
||||
name: str | None = None
|
||||
site_detail_url: str | None = None
|
||||
|
||||
def __post_init__(self):
|
||||
if isinstance(self.aliases, str):
|
||||
self.aliases = [alias for alias in self.aliases.split("\n")]
|
||||
if isinstance(self.date_added, str):
|
||||
self.date_added = datetime.strptime(self.date_added, "%Y-%m-%d %H:%M:%S")
|
||||
if isinstance(self.date_last_updated, str):
|
||||
self.date_last_updated = datetime.strptime(
|
||||
self.date_last_updated, "%Y-%m-%d %H:%M:%S"
|
||||
)
|
||||
if isinstance(self.characters, str):
|
||||
self.characters = [
|
||||
Character(**character) for character in self.characters.split("\n")
|
||||
]
|
||||
if isinstance(self.characters, list):
|
||||
self.characters = [Character(**character) for character in self.characters]
|
||||
|
||||
|
||||
@dataclass
|
||||
class Producer:
|
||||
api_detail_url: str = None
|
||||
id: int = None
|
||||
name: str = None
|
||||
site_detail_url: str = None
|
||||
|
||||
|
||||
@dataclass
|
||||
class Promo:
|
||||
api_detail_url: str | None = None
|
||||
date_added: datetime | str | None = None
|
||||
deck: str | None = None
|
||||
id: int | None = None
|
||||
image: Image | str | None = None
|
||||
link: str | None = None
|
||||
name: str | None = None
|
||||
resource_type: str | None = None
|
||||
user: str | None = None
|
||||
guid: str | None = None
|
||||
|
||||
def __str__(self):
|
||||
return self.name
|
||||
|
||||
|
||||
@dataclass
|
||||
class Publisher:
|
||||
aliases: str | list | None = None
|
||||
api_detail_url: str | None = None
|
||||
characters: str | list | None = None
|
||||
date_added: datetime | str | None = None
|
||||
date_last_updated: datetime | str | None = None
|
||||
deck: str | None = None
|
||||
description: str | None = None
|
||||
id: int | None = None
|
||||
image: Image | str | None = None
|
||||
location_address: str | None = None
|
||||
location_city: str | None = None
|
||||
location_state: str | None = None
|
||||
name: str | None = None
|
||||
site_detail_url: str | None = None
|
||||
story_arcs: str | list | None = None
|
||||
teams: str | list | None = None
|
||||
volumes: str | list | None = None
|
||||
|
||||
def __post_init__(self):
|
||||
if isinstance(self.aliases, str):
|
||||
self.aliases = [alias for alias in self.aliases.split("\n")]
|
||||
if isinstance(self.image, str):
|
||||
self.image = Image(**self.image)
|
||||
if isinstance(self.date_added, str):
|
||||
self.date_added = datetime.strptime(self.date_added, "%Y-%m-%d %H:%M:%S")
|
||||
if isinstance(self.date_last_updated, str):
|
||||
self.date_last_updated = datetime.strptime(
|
||||
self.date_last_updated, "%Y-%m-%d %H:%M:%S"
|
||||
)
|
||||
|
||||
|
||||
@dataclass
|
||||
class StaffReview:
|
||||
api_detail_url: str | None = None
|
||||
id: int | None = None
|
||||
name: str | None = None
|
||||
site_detail_url: str | None = None
|
||||
|
||||
|
||||
@dataclass
|
||||
class Series:
|
||||
aliases: str | None = None
|
||||
api_detail_url: str | None = None
|
||||
character_credits: str | None = None
|
||||
count_of_episodes: int | None = None
|
||||
date_added: datetime | str | None = None
|
||||
date_last_updated: datetime | str | None = None
|
||||
deck: str | None = None
|
||||
description: str | None = None
|
||||
first_episode: str | None = None
|
||||
id: int | None = None
|
||||
image: Image | str | None = None
|
||||
last_episode: str | None = None
|
||||
location_credits: list | None = None
|
||||
name: str | None = None
|
||||
publisher: dict | None = None
|
||||
site_detail_url: str | None = None
|
||||
start_year: int | None = None
|
||||
|
||||
def __post_init__(self):
|
||||
if isinstance(self.aliases, str):
|
||||
self.aliases = [alias for alias in self.aliases.split("\n")]
|
||||
if isinstance(self.date_added, str):
|
||||
self.date_added = datetime.strptime(self.date_added, "%Y-%m-%d %H:%M:%S")
|
||||
if isinstance(self.date_last_updated, str):
|
||||
self.date_last_updated = datetime.strptime(
|
||||
self.date_last_updated, "%Y-%m-%d %H:%M:%S"
|
||||
)
|
||||
if isinstance(self.image, dict):
|
||||
self.image = Image(**self.image)
|
||||
if isinstance(self.location_credits, list):
|
||||
self.location_credits = [
|
||||
Location(**location) for location in self.location_credits
|
||||
]
|
||||
log.debug(self.publisher)
|
||||
if isinstance(self.publisher, dict):
|
||||
self.publisher = Publisher(**self.publisher)
|
||||
|
||||
|
||||
@dataclass
|
||||
class StoryArcs:
|
||||
aliases: str | list | None = None
|
||||
api_detail_url: str | None = None
|
||||
count_of_isssue_appearances: int | None = None
|
||||
date_added: datetime | str | None = None
|
||||
date_last_updated: datetime | str | None = None
|
||||
deck: str | None = None
|
||||
description: str | None = None
|
||||
first_appeared_in_issue: str | None = None
|
||||
id: int | None = None
|
||||
image: Image | dict | None = None
|
||||
issues: str | None = None
|
||||
first_appeared_in_episode: list | None = None
|
||||
name: str | None = None
|
||||
publisher: list | None = None
|
||||
site_detail_url: str | None = None
|
||||
|
||||
def __post_init__(self):
|
||||
if isinstance(self.aliases, str):
|
||||
self.aliases = [alias for alias in self.aliases.split("\n")]
|
||||
if isinstance(self.date_added, str):
|
||||
self.date_added = datetime.strptime(self.date_added, "%Y-%m-%d %H:%M:%S")
|
||||
if isinstance(self.date_last_updated, str):
|
||||
self.date_last_updated = datetime.strptime(
|
||||
self.date_last_updated, "%Y-%m-%d %H:%M:%S"
|
||||
)
|
||||
if isinstance(self.image, dict):
|
||||
self.image = Image(**self.image)
|
||||
if isinstance(self.first_appeared_in_episode, list):
|
||||
self.first_appeared_in_episode = [
|
||||
Episode(**episode) for episode in self.first_appeared_in_episode
|
||||
]
|
||||
if isinstance(self.publisher, list):
|
||||
self.publisher = [Publisher(**publisher) for publisher in self.publisher]
|
||||
|
||||
|
||||
@dataclass
|
||||
class StoryArc(StoryArcs):
|
||||
episodes: list | None = None
|
||||
movies: list | None = None
|
||||
|
||||
def __post_init__(self):
|
||||
if isinstance(self.episodes, list):
|
||||
self.episodes = [Episode(**episode) for episode in self.episodes]
|
||||
if isinstance(self.movies, list):
|
||||
self.movies = [Movie(**movie) for movie in self.movies]
|
||||
|
||||
|
||||
@dataclass
|
||||
class Studio:
|
||||
api_detail_url: str | None = None
|
||||
id: int | None = None
|
||||
name: str | None = None
|
||||
site_detail_url: str | None = None
|
||||
|
||||
|
||||
@dataclass
|
||||
class Teams:
|
||||
aliases: str | list | None = None
|
||||
api_detail_url: str | None = None
|
||||
character_enemies: list[Character] | list | None = None
|
||||
character_friends: list[Character] | list | None = None
|
||||
characters: list[Character] | list | None = None
|
||||
count_of_isssue_appearances: int | None = None
|
||||
count_team_members: int | None = None
|
||||
date_added: datetime | str | None = None
|
||||
date_last_updated: datetime | str | None = None
|
||||
deck: str | None = None
|
||||
description: str | None = None
|
||||
disbanded_in_issues: list[Issue] | None = None
|
||||
first_appeared_in_issue: str | None = None
|
||||
id: int | None = None
|
||||
image: Image | dict | None = None
|
||||
issue_credits: list | None = None
|
||||
issues_disbanded_in: list[Issue] | None = None
|
||||
movies: list[Movie] | list | None = None
|
||||
name: str | None = None
|
||||
publisher: Publisher | None = None
|
||||
site_detail_url: str | None = None
|
||||
story_arc_credits: list | None = None
|
||||
volume_credits: list | None = None
|
||||
count_of_team_members: int | None = None
|
||||
|
||||
def __post_init__(self):
|
||||
if isinstance(self.aliases, str):
|
||||
self.aliases = [alias for alias in self.aliases.split("\n")]
|
||||
if isinstance(self.character_enemies, list):
|
||||
self.character_enemies = [
|
||||
Character(**character) for character in self.character_enemies
|
||||
]
|
||||
if isinstance(self.character_friends, list):
|
||||
self.character_friends = [
|
||||
Character(**character) for character in self.character_friends
|
||||
]
|
||||
if isinstance(self.characters, list):
|
||||
self.characters = [Character(**character) for character in self.characters]
|
||||
if isinstance(self.disbanded_in_issues, list):
|
||||
self.disbanded_in_issues = [
|
||||
Issue(**issue) for issue in self.disbanded_in_issues
|
||||
]
|
||||
if isinstance(self.image, dict):
|
||||
self.image = Image(**self.image)
|
||||
if isinstance(self.date_added, str):
|
||||
self.date_added = datetime.strptime(self.date_added, "%Y-%m-%d %H:%M:%S")
|
||||
if isinstance(self.date_last_updated, str):
|
||||
self.date_last_updated = datetime.strptime(
|
||||
self.date_last_updated, "%Y-%m-%d %H:%M:%S"
|
||||
)
|
||||
if isinstance(self.issues_disbanded_in, list):
|
||||
self.issues_disbanded_in = [
|
||||
Issue(**issue) for issue in self.issues_disbanded_in
|
||||
]
|
||||
|
||||
|
||||
@dataclass
|
||||
class Team(Teams):
|
||||
characters: list | None = None
|
||||
character_enemies: list | None = None
|
||||
character_friends: list | None = None
|
||||
isssues_disbanded_in: list | None = None
|
||||
|
||||
def __post_init__(self):
|
||||
if isinstance(self.characters, list):
|
||||
self.characters = [Character(**character) for character in self.characters]
|
||||
if isinstance(self.character_enemies, list):
|
||||
self.character_enemies = [
|
||||
Character(**character) for character in self.character_enemies
|
||||
]
|
||||
if isinstance(self.character_friends, list):
|
||||
self.character_friends = [
|
||||
Character(**character) for character in self.character_friends
|
||||
]
|
||||
if isinstance(self.isssues_disbanded_in, list):
|
||||
self.isssues_disbanded_in = [
|
||||
Issue(**issue) for issue in self.isssues_disbanded_in
|
||||
]
|
||||
|
||||
|
||||
@dataclass
|
||||
class Type:
|
||||
detail_resource_name: str | None = None
|
||||
id: int | None = None
|
||||
list_resource_name: str | None = None
|
||||
|
||||
|
||||
@dataclass
|
||||
class Video:
|
||||
api_detail_url: str | None = None
|
||||
deck: str | None = None
|
||||
hd_url: str | None = None
|
||||
id: int | None = None
|
||||
image: Image | dict | None = None
|
||||
length_seconds: int | None = None
|
||||
low_url: str | None = None
|
||||
publish_date: datetime | str | None = None
|
||||
site_detail_url: str | None = None
|
||||
url: str | None = None
|
||||
user: str | None = None
|
||||
|
||||
def __post_init__(self):
|
||||
if isinstance(self.image, dict):
|
||||
self.image = Image(**self.image)
|
||||
if isinstance(self.publish_date, str):
|
||||
self.publish_date = datetime.strptime(
|
||||
self.publish_date, "%Y-%m-%d %H:%M:%S"
|
||||
)
|
||||
|
||||
|
||||
@dataclass
|
||||
class VideoType:
|
||||
api_detail_url: str | None = None
|
||||
deck: str | None = None
|
||||
id: int | None = None
|
||||
name: str | None = None
|
||||
site_detail_url: str | None = None
|
||||
|
||||
|
||||
@dataclass
|
||||
class VideoCategory:
|
||||
api_detail_url: str | None = None
|
||||
deck: str | None = None
|
||||
id: int | None = None
|
||||
name: str | None = None
|
||||
site_detail_url: str | None = None
|
||||
|
||||
|
||||
#
|
||||
@dataclass
|
||||
class Volumes:
|
||||
aliases: str | None = None
|
||||
api_detail_url: str | None = None
|
||||
count_of_issues: int | None = None
|
||||
date_added: str | None = None
|
||||
date_last_updated: str | None = None
|
||||
deck: str | None = None
|
||||
description: str | None = None
|
||||
first_issue: dict | None = None
|
||||
id: int | None = None
|
||||
image: Image | None = None
|
||||
last_issue: dict | None = None
|
||||
name: str | None = None
|
||||
publisher: dict | None = None
|
||||
site_detail_url: str | None = None
|
||||
start_year: int | None = None
|
||||
|
||||
def __post_init__(self):
|
||||
if isinstance(self.publisher, dict):
|
||||
self.publisher = Publisher(**self.publisher)
|
||||
if isinstance(self.image, dict):
|
||||
self.image = Image(**self.image)
|
||||
|
||||
@dataclass
|
||||
class Volume(Volumes):
|
||||
characters: list | None = None
|
||||
concepts: list | None = None
|
||||
issues: list | None = None
|
||||
locations:list | None = None
|
||||
objects: list | None = None
|
||||
people:list| None = None
|
||||
def __post_init__(self):
|
||||
if isinstance(self.characters, list):
|
||||
self.characters = [Character(**character) for character in self.characters]
|
||||
if isinstance(self.concepts, list):
|
||||
self.concepts = [Concept(**concept) for concept in self.concepts]
|
||||
if isinstance(self.issues, list):
|
||||
self.issues = [Issue(**issue) for issue in self.issues]
|
||||
if isinstance(self.locations, list):
|
||||
self.locations = [Location(**location) for location in self.locations]
|
||||
if isinstance(self.objects, list):
|
||||
self.objects = [Object(**object) for object in self.objects]
|
||||
if isinstance(self.people, list):
|
||||
self.people = [Person(**person) for person in self.people]
|
||||
|
||||
|
||||
@dataclass
|
||||
class Writer:
|
||||
api_detail_url: str | None = None
|
||||
id: int | None = None
|
||||
name: str | None = None
|
||||
site_detail_url: str | None = None
|
||||
64
src/comicvineapi/schemas/response.py
Normal file
64
src/comicvineapi/schemas/response.py
Normal file
@@ -0,0 +1,64 @@
|
||||
from dataclasses import dataclass, field
|
||||
from loguru import logger
|
||||
from datetime import datetime
|
||||
|
||||
log = logger
|
||||
log.remove()
|
||||
log.add(
|
||||
f"logs/{datetime.now().strftime('%Y-%m-%d')}.log",
|
||||
rotation="1 day",
|
||||
compression="zip",
|
||||
)
|
||||
log.add("logs/api.log", compression="zip", rotation="50 MB")
|
||||
|
||||
|
||||
@dataclass
|
||||
class ComicVineResponse:
|
||||
status_code: int = 0
|
||||
error: str | None = None
|
||||
number_of_total_results: str | None = None
|
||||
number_of_page_results: str | None = None
|
||||
limit: str | None = None
|
||||
offset: str | None = None
|
||||
results: list = field(default_factory=list)
|
||||
version: float | None = None
|
||||
|
||||
def __post_init__(self):
|
||||
self.status = self.translate_status_code()
|
||||
log.debug(
|
||||
f"Response status: {self.status}, got {self.number_of_page_results} pages of {self.number_of_total_results} total results"
|
||||
)
|
||||
|
||||
def translate_status_code(self):
|
||||
match self.status_code:
|
||||
case 1:
|
||||
return "OK"
|
||||
case 100:
|
||||
raise Exception("Invalid API Key")
|
||||
case 101:
|
||||
raise Exception("Object Not Found")
|
||||
case 102:
|
||||
raise Exception("Error in URL Format")
|
||||
case 103:
|
||||
raise Exception("'jsonp' format requires a 'json_callback' argument")
|
||||
case 104:
|
||||
raise Exception("Filter Error")
|
||||
case 105:
|
||||
raise Exception("Subscriber only video is for subscribers only")
|
||||
|
||||
def handle_result(self, resultType):
|
||||
log.debug(f"Handling result, result is: {type(self.results)}")
|
||||
if self.number_of_page_results == 0:
|
||||
return [None]
|
||||
if self.number_of_total_results == 1:
|
||||
return [resultType(**self.results[0])]
|
||||
results = [None] * len(self.results)
|
||||
for i, result in enumerate(self.results):
|
||||
if isinstance(result, list):
|
||||
continue
|
||||
else:
|
||||
results[i] = resultType(**result)
|
||||
# remove any None values
|
||||
results = [x for x in results if x is not None]
|
||||
log.debug(f"Returning {len(results)} result(s)")
|
||||
return results
|
||||
28
src/comicvineapi/schemas/volume.py
Normal file
28
src/comicvineapi/schemas/volume.py
Normal file
@@ -0,0 +1,28 @@
|
||||
from dataclasses import dataclass
|
||||
from typing import List, Any
|
||||
@dataclass
|
||||
class Volume:
|
||||
aliases: str
|
||||
api_detail_url: str
|
||||
character_credits: List[Any]
|
||||
convept_credits: List[Any]
|
||||
countof_issues: int
|
||||
date_added: str
|
||||
date_last_updated: str
|
||||
deck: str
|
||||
description: str
|
||||
firts_issue: Any
|
||||
id: int
|
||||
image: Any
|
||||
last_issue: Any
|
||||
location_credits: List[Any]
|
||||
name: str
|
||||
object_credits: List[Any]
|
||||
person_credits: List[Any]
|
||||
publisher: Any
|
||||
site_detail_url: str
|
||||
start_year: int
|
||||
team_credits: List[Any]
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user