"""

"""

"""

"""

"""

"""

"""

"""

"""

def __init__(self, host, meta):

"""

# Server host name

self.host = host

self.server_info = meta

# Version from server is major.minor.rev or major.minor.rev."Dev"

# Store version as tuple and check dev flag

try:

self.version = meta.get("version", None)

except AttributeError:

self.version = None

if not self.version:

raise ShotgunError("The ShotGrid Server didn't respond with a version number. "

"This may be because you are running an older version of "

"ShotGrid against a more recent version of the ShotGrid API. "

"For more information, please contact ShotGrid Support.")

if len(self.version) > 3 and self.version[3] == "Dev":

self.is_dev = True

else:

self.is_dev = False

self.version = tuple(self.version[:3])

self._ensure_json_supported()

def _ensure_support(self, feature, raise_hell=True):

"""

if not self.version or self.version < feature["version"]:

if raise_hell:

raise ShotgunError(

"{} requires server version {} or higher, server is {}".format(feature["label"],

_version_str(feature["version"]),

_version_str(self.version))

)

return False

else:

return True

def _ensure_json_supported(self):

"""

self._ensure_support({

"version": (2, 4, 0),

"label": "JSON API"

})

def ensure_include_archived_projects(self):

"""

self._ensure_support({

"version": (5, 3, 14),

"label": "include_archived_projects parameter"

})

def ensure_per_project_customization(self):

"""

return self._ensure_support({

"version": (5, 4, 4),

"label": "project parameter"

}, True)

def ensure_support_for_additional_filter_presets(self):

"""

return self._ensure_support({

"version": (7, 0, 0),

"label": "additional_filter_presets parameter"

}, True)

def ensure_user_following_support(self):

"""

return self._ensure_support({

"version": (7, 0, 12),

"label": "user_following parameter"

}, True)

def ensure_paging_info_without_counts_support(self):

"""

return self._ensure_support({

"version": (7, 4, 0),

"label": "optimized pagination"

}, False)

def ensure_return_image_urls_support(self):

"""

return self._ensure_support({

"version": (3, 3, 0),

"label": "return thumbnail URLs"

}, False)

def __str__(self):

"""

def __init__(self):

system = sys.platform.lower()

if system == "darwin":

self.platform = "mac"

elif system.startswith("linux"):

self.platform = "linux"

elif system == "win32":

self.platform = "windows"

else:

self.platform = None

if self.platform:

self.local_path_field = f"local_path_{self.platform}"

else:

self.local_path_field = None

self.py_version = ".".join(str(x) for x in sys.version_info[:2])

# extract the OpenSSL version if we can. The version is only available in Python 2.7 and

# only if we successfully imported ssl

self.ssl_version = "unknown"

try:

self.ssl_version = ssl.OPENSSL_VERSION

except (AttributeError, NameError):

pass

def __str__(self):

return ("ClientCapabilities: platform {}, local_path_field {}, "

"py_verison {}, ssl version {}").format(self.platform,

self.local_path_field,

self.py_version,

"""

def __init__(self, sg):

"""

self._sg = sg

self.max_rpc_attempts = 3

# rpc_attempt_interval stores the number of milliseconds to wait between

# request retries. By default, this will be 3000 milliseconds. You can

# override this by setting this property on the config like so:

#

# sg = Shotgun(site_name, script_name, script_key)

# sg.config.rpc_attempt_interval = 1000 # adjusting default interval

#

# Or by setting the ``SHOTGUN_API_RETRY_INTERVAL`` environment variable.

# In the case that the environment variable is already set, setting the

# property on the config will override it.

self.rpc_attempt_interval = 3000

# From http://docs.python.org/2.6/library/httplib.html:

# If the optional timeout parameter is given, blocking operations

# (like connection attempts) will timeout after that many seconds

# (if it is not given, the global default timeout setting is used)

self.timeout_secs = None

self.api_ver = "api3"

self.convert_datetimes_to_utc = True

self._records_per_page = None

self.api_key = None

self.script_name = None

self.user_login = None

self.user_password = None

self.auth_token = None

self.sudo_as_login = None

# Authentication parameters to be folded into final auth_params dict

self.extra_auth_params = None

# uuid as a string

self.session_uuid = None

self.scheme = None

self.server = None

self.api_path = None

# The raw_http_proxy reflects the exact string passed in

# to the Shotgun constructor. This can be useful if you

# need to construct a Shotgun API instance based on

# another Shotgun API instance.

self.raw_http_proxy = None

# if a proxy server is being used, the proxy_handler

# below will contain a urllib2.ProxyHandler instance

# which can be used whenever a request needs to be made.

self.proxy_handler = None

self.proxy_server = None

self.proxy_port = 8080

self.proxy_user = None

self.proxy_pass = None

self.session_token = None

self.authorization = None

self.no_ssl_validation = False

self.localized = False

def set_server_params(self, base_url):

"""

self.scheme, self.server, api_base, _, _ = \

urllib.parse.urlsplit(base_url)

if self.scheme not in ("http", "https"):

raise ValueError(

f"base_url must use http or https got '{base_url}'"

)

self.api_path = urllib.parse.urljoin(urllib.parse.urljoin(

api_base or "/", self.api_ver + "/"), "json"

)

@property

def records_per_page(self):

"""

if self._records_per_page is None:

# Check for api_max_entities_per_page in the server info and change the record per page

# value if it is supplied.

self._records_per_page = self._sg.server_info.get("api_max_entities_per_page") or 500

"""

# reg ex from

# http://underground.infovark.com/2008/07/22/iso-date-validation-regex/

# Note a length check is done before checking the reg ex

_DATE_PATTERN = re.compile(

r"^(\d{4})\D?(0[1-9]|1[0-2])\D?([12]\d|0[1-9]|3[01])$")

_DATE_TIME_PATTERN = re.compile(

r"^(\d{4})\D?(0[1-9]|1[0-2])\D?([12]\d|0[1-9]|3[01])"

r"(\D?([01]\d|2[0-3])\D?([0-5]\d)\D?([0-5]\d)?\D?(\d{3})?)?$")

_MULTIPART_UPLOAD_CHUNK_SIZE = 20000000

def __init__(self,

base_url,

script_name=None,

api_key=None,

convert_datetimes_to_utc=True,

http_proxy=None,

ensure_ascii=True,

connect=True,

ca_certs=None,

login=None,

password=None,

sudo_as_login=None,

session_token=None,

auth_token=None):

"""

# verify authentication arguments

if session_token is not None:

if script_name is not None or api_key is not None:

raise ValueError("cannot provide both session_token "

"and script_name/api_key")

if login is not None or password is not None:

raise ValueError("cannot provide both session_token "

"and login/password")

if login is not None or password is not None:

if script_name is not None or api_key is not None:

raise ValueError("cannot provide both login/password "

"and script_name/api_key")

if login is None:

raise ValueError("password provided without login")

if password is None:

raise ValueError("login provided without password")

if script_name is not None or api_key is not None:

if script_name is None:

raise ValueError("api_key provided without script_name")

if api_key is None:

raise ValueError("script_name provided without api_key")

if auth_token is not None:

if login is None or password is None:

raise ValueError("must provide a user login and password with an auth_token")

if script_name is not None or api_key is not None:

raise ValueError("cannot provide an auth_code with script_name/api_key")

if all([session_token, script_name, api_key, login, password]):

if connect:

raise ValueError("must provide login/password, session_token or script_name/api_key")

self.config = _Config(self)

self.config.api_key = api_key

self.config.script_name = script_name

self.config.user_login = login

self.config.user_password = password

self.config.auth_token = auth_token

self.config.session_token = session_token

self.config.sudo_as_login = sudo_as_login

self.config.convert_datetimes_to_utc = convert_datetimes_to_utc

self.config.no_ssl_validation = NO_SSL_VALIDATION

self.config.raw_http_proxy = http_proxy

try:

self.config.rpc_attempt_interval = int(os.environ.get("SHOTGUN_API_RETRY_INTERVAL", 3000))

except ValueError:

retry_interval = os.environ.get("SHOTGUN_API_RETRY_INTERVAL", 3000)

raise ValueError(f"Invalid value '{retry_interval}' found in environment variable "

"SHOTGUN_API_RETRY_INTERVAL, must be int.")

if self.config.rpc_attempt_interval < 0:

raise ValueError("Value of SHOTGUN_API_RETRY_INTERVAL must be positive, "

f"got '{self.config.rpc_attempt_interval}'.")

self._connection = None

self.__ca_certs = self._get_certs_file(ca_certs)

self.base_url = (base_url or "").lower()

self.config.set_server_params(self.base_url)

# if the service contains user information strip it out

# copied from the xmlrpclib which turned the user:password into

# and auth header

# Do NOT self._split_url(self.base_url) here, as it contains the lower

# case version of the base_url argument. Doing so would base64encode

# the lowercase version of the credentials.

auth, self.config.server = self._split_url(base_url)

if auth:

auth = base64encode(urllib.parse.unquote(auth).encode(ENCODING)).decode(ENCODING)

self.config.authorization = "Basic " + auth.strip()

# foo:[email protected]:3456

if http_proxy:

# check if we're using authentication. Start from the end since

# there might be @ in the user's password.

p = http_proxy.rsplit("@", 1)

if len(p) > 1:

self.config.proxy_user, self.config.proxy_pass = \

p[0].split(":", 1)

proxy_server = p[1]

else:

proxy_server = http_proxy

proxy_netloc_list = proxy_server.split(":", 1)

self.config.proxy_server = proxy_netloc_list[0]

if len(proxy_netloc_list) > 1:

try:

self.config.proxy_port = int(proxy_netloc_list[1])

except ValueError:

raise ValueError(f"Invalid http_proxy address '{http_proxy}'. Valid "

"format is '123.456.789.012' or '123.456.789.012:3456'"

f". If no port is specified, a default of {self.config.proxy_port} will be "

"used.")

# now populate self.config.proxy_handler

if self.config.proxy_user and self.config.proxy_pass:

auth_string = f"{self.config.proxy_user}:{self.config.proxy_pass}@"

else:

auth_string = ""

proxy_addr = f"http://{auth_string}{self.config.proxy_server}:{self.config.proxy_port}"

self.config.proxy_handler = urllib.request.ProxyHandler({self.config.scheme: proxy_addr})

if ensure_ascii:

self._json_loads = self._json_loads_ascii

self.client_caps = ClientCapabilities()

# this relies on self.client_caps being set first

self.reset_user_agent()

self._server_caps = None

# test to ensure the the server supports the json API

# call to server will only be made once and will raise error

if connect:

self.server_caps

# When using auth_token in a 2FA scenario we need to switch to session-based

# authentication because the auth token will no longer be valid after a first use.

if self.config.auth_token is not None:

self.config.session_token = self.get_session_token()

self.config.user_login = None

self.config.user_password = None

self.config.auth_token = None

def _split_url(self, base_url):

"""

auth = None

results = urllib.parse.urlparse(base_url)

server = results.hostname

if results.port:

server = f"{server}:{results.port}"

if results.username:

auth = results.username

if results.password:

auth = f"{auth}:{results.password}"

return auth, server

# ========================================================================

# API Functions

@property

def server_info(self):

"""

return self.server_caps.server_info

@property

def server_caps(self):

"""

if not self._server_caps or (self._server_caps.host != self.config.server):

self._server_caps = ServerCapabilities(self.config.server, self.info())

return self._server_caps

def connect(self):

"""

self._get_connection()

self.info()

return

def close(self):

"""

self._close_connection()

return

def info(self):

"""

return self._call_rpc("info", None, include_auth_params=False)

def find_one(self, entity_type, filters, fields=None, order=None, filter_operator=None, retired_only=False,

include_archived_projects=True, additional_filter_presets=None):

"""

results = self.find(entity_type, filters, fields, order, filter_operator, 1, retired_only,

include_archived_projects=include_archived_projects,

additional_filter_presets=additional_filter_presets)

if results:

return results[0]

return None

def find(self, entity_type, filters, fields=None, order=None, filter_operator=None, limit=0,

retired_only=False, page=0, include_archived_projects=True, additional_filter_presets=None):

"""

if not isinstance(limit, int) or limit < 0:

raise ValueError("limit parameter must be a positive integer")

if not isinstance(page, int) or page < 0:

raise ValueError("page parameter must be a positive integer")

if isinstance(filters, (list, tuple)):

filters = _translate_filters(filters, filter_operator)

elif filter_operator:

# TODO: Not sure if this test is correct, replicated from prev api

raise ShotgunError("Deprecated: Use of filter_operator for find() is not valid any more."

" See the documentation on find()")

if not include_archived_projects:

# This defaults to True on the server (no argument is sent)

# So we only need to check the server version if it is False

self.server_caps.ensure_include_archived_projects()

if additional_filter_presets:

self.server_caps.ensure_support_for_additional_filter_presets()

params = self._construct_read_parameters(entity_type,

fields,

filters,

retired_only,

order,

include_archived_projects,

additional_filter_presets)

if self.server_caps.ensure_return_image_urls_support():

params["api_return_image_urls"] = True

if self.server_caps.ensure_paging_info_without_counts_support():

paging_info_param = "return_paging_info_without_counts"

else:

paging_info_param = "return_paging_info"

params[paging_info_param] = False