File: base.py

package info (click to toggle)
python-tmdbsimple 2.9.1-1
  • links: PTS, VCS
  • area: main
  • in suites: bookworm, forky, sid, trixie
  • size: 348 kB
  • sloc: python: 2,231; makefile: 4
file content (131 lines) | stat: -rw-r--r-- 4,160 bytes parent folder | download
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
# -*- coding: utf-8 -*-

"""
tmdbsimple.base
~~~~~~~~~~~~~~~
This module implements the base class of tmdbsimple.

Created by Celia Oakley on 2013-10-31.

:copyright: (c) 2013-2022 by Celia Oakley
:license: GPLv3, see LICENSE for more details
"""

import json
import requests


class APIKeyError(Exception):
    pass


class TMDB(object):
    headers = {'Content-Type': 'application/json',
               'Accept': 'application/json',
               'Connection': 'close'}
    BASE_PATH = ''
    URLS = {}

    def __init__(self):
        from . import API_VERSION, REQUESTS_SESSION, REQUESTS_TIMEOUT
        self.base_uri = 'https://api.themoviedb.org'
        self.base_uri += '/{version}'.format(version=API_VERSION)
        self.session = REQUESTS_SESSION
        self.timeout = REQUESTS_TIMEOUT

    def _get_path(self, key):
        return self.BASE_PATH + self.URLS[key]

    def _get_id_path(self, key):
        return self._get_path(key).format(id=self.id)

    def _get_guest_session_id_path(self, key):
        return self._get_path(key).format(
            guest_session_id=self.guest_session_id)

    def _get_credit_id_path(self, key):
        return self._get_path(key).format(credit_id=self.credit_id)

    def _get_media_type_time_window_path(self, key):
        return self._get_path(key).format(
            media_type=self.media_type, time_window=self.time_window)

    def _get_tv_id_season_number_path(self, key):
        return self._get_path(key).format(
            tv_id=self.tv_id, season_number=self.season_number)

    def _get_tv_id_season_number_episode_number_path(self, key):
        return self._get_path(key).format(
            tv_id=self.tv_id, season_number=self.season_number,
            episode_number=self.episode_number)

    def _get_complete_url(self, path):
        return '{base_uri}/{path}'.format(base_uri=self.base_uri, path=path)

    def _get_params(self, params):
        from . import API_KEY
        if not API_KEY:
            raise APIKeyError

        api_dict = {'api_key': API_KEY}
        if params:
            params.update(api_dict)
            for key, value in params.items():
                if isinstance(params[key], bool):
                    params[key] = 'true' if value is True else 'false'

        else:
            params = api_dict
        return params

    def _request(self, method, path, params=None, payload=None):
        url = self._get_complete_url(path)
        params = self._get_params(params)

        # Create a new request session if no global session is defined
        if self.session is None:
            response = requests.request(
                method,
                url,
                params=params,
                data=json.dumps(payload) if payload else payload,
                headers=self.headers, timeout=self.timeout
            )

        # Use the global requests session the user provided
        else:
            response = self.session.request(
                method,
                url,
                params=params,
                data=json.dumps(payload) if payload else payload,
                headers=self.headers, timeout=self.timeout
            )

        response.raise_for_status()
        response.encoding = 'utf-8'
        return response.json()

    def _GET(self, path, params=None):
        return self._request('GET', path, params=params)

    def _POST(self, path, params=None, payload=None):
        return self._request('POST', path, params=params, payload=payload)

    def _DELETE(self, path, params=None, payload=None):
        return self._request('DELETE', path, params=params, payload=payload)

    def _set_attrs_to_values(self, response={}):
        """
        Set attributes to dictionary values.

        - e.g.
        >>> import tmdbsimple as tmdb
        >>> movie = tmdb.Movies(103332)
        >>> response = movie.info()
        >>> movie.title  # instead of response['title']
        """
        if isinstance(response, dict):
            for key in response.keys():
                if not hasattr(self, key) or not callable(getattr(self, key)):
                    setattr(self, key, response[key])