#12892

pending completion

Build # #12892

Build Type

push

travis-ci

Committed by

web-flow

Commit Message

<a href="https://github.com/datajoint/datajoint-python/commit/<a class=hub.com/datajoint/datajoint-python/commit/af666718cc281d8f1183aa91fa17a2b019666abd">af666718c<a href="https://github.com/datajoint/datajoint-python/commit/af666718cc281d8f1183aa91fa17a2b019666abd">">Merge </a><a class="double-link" href="https://github.com/datajoint/datajoint-python/commit/<a class="double-link" href="https://github.com/datajoint/datajoint-python/commit/45fb43e6aa8f80700c2ee554361ddc80893ad7fa">45fb43e6a</a>">45fb43e6a</a><a href="https://github.com/datajoint/datajoint-python/commit/af666718cc281d8f1183aa91fa17a2b019666abd"> into 3b6e84588">3b6e84588</a>

Run Details

8 of 8 new or added lines in 1 file covered. (100.0%)

3035 of 3359 relevant lines covered (90.35%)

0.9 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

94.12

/datajoint/jobs.py

import os
from .hash import key_hash
import platform
from .table import Table
from .settings import config
from .errors import DuplicateError
from .heading import Heading

ERROR_MESSAGE_LENGTH = 2047
TRUNCATION_APPENDIX = "...truncated"


class JobTable(Table):
    """
    A base table with no definition. Allows reserving jobs
    """

    def __init__(self, conn, database):
        self.database = database
        self._connection = conn
        self._heading = Heading(
            table_info=dict(
                conn=conn, database=database, table_name=self.table_name, context=None
            )
        )
        self._support = [self.full_table_name]

        self._definition = """    # job reservation table for `{database}`
        table_name  :varchar(255)  # className of the table
        key_hash  :char(32)  # key hash
        ---
        status  :enum('reserved','error','ignore')  # if tuple is missing, the job is available
        key=null  :blob  # structure containing the key
        error_message=""  :varchar({error_message_length})  # error message returned if failed
        error_stack=null  :mediumblob  # error stack if failed
        user="" :varchar(255) # database user
        host=""  :varchar(255)  # system hostname
        pid=0  :int unsigned  # system process id
        connection_id = 0  : bigint unsigned          # connection_id()
        timestamp=CURRENT_TIMESTAMP  :timestamp   # automatic timestamp
        """.format(
            database=database, error_message_length=ERROR_MESSAGE_LENGTH
        )
        if not self.is_declared:
            self.declare()
        self._user = self.connection.get_user()

    @property
    def definition(self):
        return self._definition

    @property
    def table_name(self):
        return "~jobs"

    def delete(self):
        """bypass interactive prompts and dependencies"""
        self.delete_quick()

    def drop(self):
        """bypass interactive prompts and dependencies"""
        self.drop_quick()

    def reserve(self, table_name, key):
        """
        Reserve a job for computation.  When a job is reserved, the job table contains an entry for the
        job key, identified by its hash. When jobs are completed, the entry is removed.

        :param table_name: `database`.`table_name`
        :param key: the dict of the job's primary key
        :return: True if reserved job successfully. False = the jobs is already taken
        """
        job = dict(
            table_name=table_name,
            key_hash=key_hash(key),
            status="reserved",
            host=platform.node(),
            pid=os.getpid(),
            connection_id=self.connection.connection_id,
            key=key,
            user=self._user,
        )
        try:
            with config(enable_python_native_blobs=True):
                self.insert1(job, ignore_extra_fields=True)
        except DuplicateError:
            return False
        return True

    def ignore(self, table_name, key):
        """
        Set a job to be ignored for computation.  When a job is ignored, the job table contains an entry for the
        job key, identified by its hash, with status "ignore".

        :param table_name: `database`.`table_name`
        :param key: the dict of the job's primary key
        :return: True if ignore job successfully. False = the jobs is already taken
        """
        job = dict(
            table_name=table_name,
            key_hash=key_hash(key),
            status="ignore",
            host=platform.node(),
            pid=os.getpid(),
            connection_id=self.connection.connection_id,
            key=key,
            user=self._user,
        )
        try:
            with config(enable_python_native_blobs=True):
                self.insert1(job, ignore_extra_fields=True)
        except DuplicateError:
            return False
        return True

    def complete(self, table_name, key):
        """
        Log a completed job.  When a job is completed, its reservation entry is deleted.

        :param table_name: `database`.`table_name`
        :param key: the dict of the job's primary key
        """
        job_key = dict(table_name=table_name, key_hash=key_hash(key))
        (self & job_key).delete_quick()

    def error(self, table_name, key, error_message, error_stack=None):
        """
        Log an error message.  The job reservation is replaced with an error entry.
        if an error occurs, leave an entry describing the problem

        :param table_name: `database`.`table_name`
        :param key: the dict of the job's primary key
        :param error_message: string error message
        :param error_stack: stack trace
        """
        if len(error_message) > ERROR_MESSAGE_LENGTH:
            error_message = (
                error_message[: ERROR_MESSAGE_LENGTH - len(TRUNCATION_APPENDIX)]
                + TRUNCATION_APPENDIX
            )
        with config(enable_python_native_blobs=True):
            self.insert1(
                dict(
                    table_name=table_name,
                    key_hash=key_hash(key),
                    status="error",
                    host=platform.node(),
                    pid=os.getpid(),
                    connection_id=self.connection.connection_id,
                    user=self._user,
                    key=key,
                    error_message=error_message,
                    error_stack=error_stack,
                ),
                replace=True,
                ignore_extra_fields=True,
            )

1	import os	1✔
2	from .hash import key_hash	1✔
3	import platform	1✔
4	from .table import Table	1✔
5	from .settings import config	1✔
6	from .errors import DuplicateError	1✔
7	from .heading import Heading	1✔
8
9	ERROR_MESSAGE_LENGTH = 2047	1✔
10	TRUNCATION_APPENDIX = "...truncated"	1✔
11
12
13	class JobTable(Table):	1✔
14	"""
15	A base table with no definition. Allows reserving jobs
16	"""
17
18	def __init__(self, conn, database):	1✔
19	self.database = database	1✔
20	self._connection = conn	1✔
21	self._heading = Heading(	1✔
22	table_info=dict(
23	conn=conn, database=database, table_name=self.table_name, context=None
24	)
25	)
26	self._support = [self.full_table_name]	1✔
27
28	self._definition = """ # job reservation table for `{database}`	1✔
29	table_name :varchar(255) # className of the table
30	key_hash :char(32) # key hash
31	---
32	status :enum('reserved','error','ignore') # if tuple is missing, the job is available
33	key=null :blob # structure containing the key
34	error_message="" :varchar({error_message_length}) # error message returned if failed
35	error_stack=null :mediumblob # error stack if failed
36	user="" :varchar(255) # database user
37	host="" :varchar(255) # system hostname
38	pid=0 :int unsigned # system process id
39	connection_id = 0 : bigint unsigned # connection_id()
40	timestamp=CURRENT_TIMESTAMP :timestamp # automatic timestamp
41	""".format(
42	database=database, error_message_length=ERROR_MESSAGE_LENGTH
43	)
44	if not self.is_declared:	1✔
45	self.declare()	1✔
46	self._user = self.connection.get_user()	1✔
47
48	@property	1✔
49	def definition(self):
50	return self._definition	1✔
51
52	@property	1✔
53	def table_name(self):
54	return "~jobs"	1✔
55
56	def delete(self):	1✔
57	"""bypass interactive prompts and dependencies"""
58	self.delete_quick()	1✔
59
60	def drop(self):	1✔
61	"""bypass interactive prompts and dependencies"""
62	self.drop_quick()	×
63
64	def reserve(self, table_name, key):	1✔
65	"""
66	Reserve a job for computation. When a job is reserved, the job table contains an entry for the
67	job key, identified by its hash. When jobs are completed, the entry is removed.
68
69	:param table_name: `database`.`table_name`
70	:param key: the dict of the job's primary key
71	:return: True if reserved job successfully. False = the jobs is already taken
72	"""
73	job = dict(	1✔
74	table_name=table_name,
75	key_hash=key_hash(key),
76	status="reserved",
77	host=platform.node(),
78	pid=os.getpid(),
79	connection_id=self.connection.connection_id,
80	key=key,
81	user=self._user,
82	)
83	try:	1✔
84	with config(enable_python_native_blobs=True):	1✔
85	self.insert1(job, ignore_extra_fields=True)	1✔
86	except DuplicateError:	1✔
87	return False	1✔
88	return True	1✔
89
90	def ignore(self, table_name, key):	1✔
91	"""
92	Set a job to be ignored for computation. When a job is ignored, the job table contains an entry for the
93	job key, identified by its hash, with status "ignore".
94
95	:param table_name: `database`.`table_name`
96	:param key: the dict of the job's primary key
97	:return: True if ignore job successfully. False = the jobs is already taken
98	"""
99	job = dict(	1✔
100	table_name=table_name,
101	key_hash=key_hash(key),
102	status="ignore",
103	host=platform.node(),
104	pid=os.getpid(),
105	connection_id=self.connection.connection_id,
106	key=key,
107	user=self._user,
108	)
109	try:	1✔
110	with config(enable_python_native_blobs=True):	1✔
111	self.insert1(job, ignore_extra_fields=True)	1✔
112	except DuplicateError:	×
113	return False	×
114	return True	1✔
115
116	def complete(self, table_name, key):	1✔
117	"""
118	Log a completed job. When a job is completed, its reservation entry is deleted.
119
120	:param table_name: `database`.`table_name`
121	:param key: the dict of the job's primary key
122	"""
123	job_key = dict(table_name=table_name, key_hash=key_hash(key))	1✔
124	(self & job_key).delete_quick()	1✔
125
126	def error(self, table_name, key, error_message, error_stack=None):	1✔
127	"""
128	Log an error message. The job reservation is replaced with an error entry.
129	if an error occurs, leave an entry describing the problem
130
131	:param table_name: `database`.`table_name`
132	:param key: the dict of the job's primary key
133	:param error_message: string error message
134	:param error_stack: stack trace
135	"""
136	if len(error_message) > ERROR_MESSAGE_LENGTH:	1✔
137	error_message = (	1✔
138	error_message[: ERROR_MESSAGE_LENGTH - len(TRUNCATION_APPENDIX)]
139	+ TRUNCATION_APPENDIX
140	)
141	with config(enable_python_native_blobs=True):	1✔
142	self.insert1(	1✔
143	dict(
144	table_name=table_name,
145	key_hash=key_hash(key),
146	status="error",
147	host=platform.node(),
148	pid=os.getpid(),
149	connection_id=self.connection.connection_id,
150	user=self._user,
151	key=key,
152	error_message=error_message,
153	error_stack=error_stack,
154	),
155	replace=True,
156	ignore_extra_fields=True,
157	)

datajoint / datajoint-python / #12892

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous