Skip to content

connections

Connections.

This module contains the database connection and operations for the tool inventory application. It includes classes for handling database operations and custom exceptions for error handling.

Database

Database connection.

Source code in src/tool_inventory/connections.py 
 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
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
class Database:
    """Database connection."""

    def __init__(self, session: Session, /) -> None:
        """Initialize database connection.

        Args:
            session: The database session.
        """
        self.session = session

    def get_tool_by_id(self, tool_id: UUID, /) -> Tool:
        """Get a tool by ID.

        Args:
            tool_id: The UUID of the tool.

        Returns:
            The tool with the specified ID.

        Raises:
            ToolNotFoundError: If the tool is not found.
        """
        statement = select(Tool).where(Tool.id == tool_id)
        result = self.session.exec(statement)
        try:
            return result.one()
        except NoResultFound as err:
            raise ToolNotFoundError(tool_id) from err

    def get_tools(self, name: str | None = None) -> list[Tool]:
        """Get tools.

        Args:
            name: The name of the tool to filter by.

        Returns:
            A list of tools.
        """
        statement = select(Tool)
        if name:
            statement = statement.where(Tool.name == name)
        result = self.session.exec(statement)
        return list(result.all())

    def search_tools(self, query: str, /) -> list[Tool]:
        """Search tools.

        Args:
            query: The search query.

        Returns:
            A list of tools.
        """
        statement = select(Tool)
        result = self.session.exec(statement)
        matches: list[tuple[int, Tool]] = []
        for tool in result.all():
            if (score := fuzz.ratio(query.lower(), tool.name.lower())) > 50:  # noqa: PLR2004
                matches.append((score, tool))  # noqa: PERF401
        return [tool for _, tool in sorted(matches, reverse=True)]

    def create_tool(self, tool: Tool, /) -> Tool:
        """Create a tool.

        Args:
            tool: The tool to create.

        Returns:
            The created tool.

        Raises:
            ToolExistsError: If the tool already exists.
        """
        Tool.model_validate(tool)
        self.session.add(tool)
        try:
            self.session.commit()
        except IntegrityError as err:
            raise ToolExistsError(tool.id) from err
        self.session.refresh(tool)
        return tool

    def update_tool(self, tool: Tool, /) -> Tool:
        """Update a tool.

        Args:
            tool: The tool to update.

        Returns:
            The updated tool.

        Raises:
            ToolNotFoundError: If the tool is not found.
        """
        Tool.model_validate(tool)
        self.session.add(tool)
        try:
            self.session.commit()
        except IntegrityError as err:
            raise ToolNotFoundError(tool.id) from err
        self.session.refresh(tool)
        return tool

    def delete_tool(self, tool_id: UUID, /) -> None:
        """Delete a tool.

        Args:
            tool_id: The UUID of the tool to delete.

        Raises:
            ToolNotFoundError: If the tool is not found.
        """
        tool = self.get_tool_by_id(tool_id)
        self.session.delete(tool)
        self.session.commit()

__init__ 

__init__(session: Session) -> None

Initialize database connection.

Parameters:

Name Type Description Default
session Session

The database session.

 required
Source code in src/tool_inventory/connections.py 
90
91
92
93
94
95
96
def __init__(self, session: Session, /) -> None:
    """Initialize database connection.

    Args:
        session: The database session.
    """
    self.session = session

create_tool 

create_tool(tool: Tool) -> Tool

Create a tool.

Parameters:

Name Type Description Default
tool Tool

The tool to create.

 required

Returns:

Type Description
Tool

The created tool.

Raises:

Type Description
ToolExistsError

If the tool already exists.
Source code in src/tool_inventory/connections.py 
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
def create_tool(self, tool: Tool, /) -> Tool:
    """Create a tool.

    Args:
        tool: The tool to create.

    Returns:
        The created tool.

    Raises:
        ToolExistsError: If the tool already exists.
    """
    Tool.model_validate(tool)
    self.session.add(tool)
    try:
        self.session.commit()
    except IntegrityError as err:
        raise ToolExistsError(tool.id) from err
    self.session.refresh(tool)
    return tool

delete_tool 

delete_tool(tool_id: UUID) -> None

Delete a tool.

Parameters:

Name Type Description Default
tool_id UUID

The UUID of the tool to delete.

 required

Raises:

Type Description
ToolNotFoundError

If the tool is not found.
Source code in src/tool_inventory/connections.py 
191
192
193
194
195
196
197
198
199
200
201
202
def delete_tool(self, tool_id: UUID, /) -> None:
    """Delete a tool.

    Args:
        tool_id: The UUID of the tool to delete.

    Raises:
        ToolNotFoundError: If the tool is not found.
    """
    tool = self.get_tool_by_id(tool_id)
    self.session.delete(tool)
    self.session.commit()

get_tool_by_id 

get_tool_by_id(tool_id: UUID) -> Tool

Get a tool by ID.

Parameters:

Name Type Description Default
tool_id UUID

The UUID of the tool.

 required

Returns:

Type Description
Tool

The tool with the specified ID.

Raises:

Type Description
ToolNotFoundError

If the tool is not found.
Source code in src/tool_inventory/connections.py 
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
def get_tool_by_id(self, tool_id: UUID, /) -> Tool:
    """Get a tool by ID.

    Args:
        tool_id: The UUID of the tool.

    Returns:
        The tool with the specified ID.

    Raises:
        ToolNotFoundError: If the tool is not found.
    """
    statement = select(Tool).where(Tool.id == tool_id)
    result = self.session.exec(statement)
    try:
        return result.one()
    except NoResultFound as err:
        raise ToolNotFoundError(tool_id) from err

get_tools 

get_tools(name: str | None = None) -> list[Tool]

Get tools.

Parameters:

Name Type Description Default
name str | None

The name of the tool to filter by.

 None

Returns:

Type Description
list[Tool]

A list of tools.
Source code in src/tool_inventory/connections.py 
117
118
119
120
121
122
123
124
125
126
127
128
129
130
def get_tools(self, name: str | None = None) -> list[Tool]:
    """Get tools.

    Args:
        name: The name of the tool to filter by.

    Returns:
        A list of tools.
    """
    statement = select(Tool)
    if name:
        statement = statement.where(Tool.name == name)
    result = self.session.exec(statement)
    return list(result.all())

search_tools 

search_tools(query: str) -> list[Tool]

Search tools.

Parameters:

Name Type Description Default
query str

The search query.

 required

Returns:

Type Description
list[Tool]

A list of tools.
Source code in src/tool_inventory/connections.py 
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
def search_tools(self, query: str, /) -> list[Tool]:
    """Search tools.

    Args:
        query: The search query.

    Returns:
        A list of tools.
    """
    statement = select(Tool)
    result = self.session.exec(statement)
    matches: list[tuple[int, Tool]] = []
    for tool in result.all():
        if (score := fuzz.ratio(query.lower(), tool.name.lower())) > 50:  # noqa: PLR2004
            matches.append((score, tool))  # noqa: PERF401
    return [tool for _, tool in sorted(matches, reverse=True)]

update_tool 

update_tool(tool: Tool) -> Tool

Update a tool.

Parameters:

Name Type Description Default
tool Tool

The tool to update.

 required

Returns:

Type Description
Tool

The updated tool.

Raises:

Type Description
ToolNotFoundError

If the tool is not found.
Source code in src/tool_inventory/connections.py 
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
def update_tool(self, tool: Tool, /) -> Tool:
    """Update a tool.

    Args:
        tool: The tool to update.

    Returns:
        The updated tool.

    Raises:
        ToolNotFoundError: If the tool is not found.
    """
    Tool.model_validate(tool)
    self.session.add(tool)
    try:
        self.session.commit()
    except IntegrityError as err:
        raise ToolNotFoundError(tool.id) from err
    self.session.refresh(tool)
    return tool

ObjectExistsError

Bases: Exception

Object exists error.

Source code in src/tool_inventory/connections.py 
61
62
63
64
65
66
67
68
69
70
71
class ObjectExistsError(Exception):
    """Object exists error."""

    def __init__(self, object_id: UUID, /) -> None:
        """Initialize object exists error.

        Args:
            object_id: The UUID of the object.
        """
        self.object_id = object_id
        self.detail = "Object already exists"

__init__ 

__init__(object_id: UUID) -> None

Initialize object exists error.

Parameters:

Name Type Description Default
object_id UUID

The UUID of the object.

 required
Source code in src/tool_inventory/connections.py 
64
65
66
67
68
69
70
71
def __init__(self, object_id: UUID, /) -> None:
    """Initialize object exists error.

    Args:
        object_id: The UUID of the object.
    """
    self.object_id = object_id
    self.detail = "Object already exists"

ObjectNotFoundError

Bases: Exception

Object not found error.

Source code in src/tool_inventory/connections.py 
35
36
37
38
39
40
41
42
43
44
45
class ObjectNotFoundError(Exception):
    """Object not found error."""

    def __init__(self, object_id: UUID, /) -> None:
        """Initialize object not found error.

        Args:
            object_id: The UUID of the object.
        """
        self.object_id = object_id
        self.detail = "Object not found"

__init__ 

__init__(object_id: UUID) -> None

Initialize object not found error.

Parameters:

Name Type Description Default
object_id UUID

The UUID of the object.

 required
Source code in src/tool_inventory/connections.py 
38
39
40
41
42
43
44
45
def __init__(self, object_id: UUID, /) -> None:
    """Initialize object not found error.

    Args:
        object_id: The UUID of the object.
    """
    self.object_id = object_id
    self.detail = "Object not found"

ToolExistsError

Bases: ObjectExistsError

Tool exists error.

Source code in src/tool_inventory/connections.py 
74
75
76
77
78
79
80
81
82
83
84
class ToolExistsError(ObjectExistsError):
    """Tool exists error."""

    def __init__(self, tool_id: UUID, /) -> None:
        """Initialize tool exists error.

        Args:
            tool_id: The UUID of the tool.
        """
        super().__init__(tool_id)
        self.detail = "Tool already exists"

__init__ 

__init__(tool_id: UUID) -> None

Initialize tool exists error.

Parameters:

Name Type Description Default
tool_id UUID

The UUID of the tool.

 required
Source code in src/tool_inventory/connections.py 
77
78
79
80
81
82
83
84
def __init__(self, tool_id: UUID, /) -> None:
    """Initialize tool exists error.

    Args:
        tool_id: The UUID of the tool.
    """
    super().__init__(tool_id)
    self.detail = "Tool already exists"

ToolNotFoundError

Bases: ObjectNotFoundError

Tool not found error.

Source code in src/tool_inventory/connections.py 
48
49
50
51
52
53
54
55
56
57
58
class ToolNotFoundError(ObjectNotFoundError):
    """Tool not found error."""

    def __init__(self, tool_id: UUID, /) -> None:
        """Initialize tool not found error.

        Args:
            tool_id: The UUID of the tool.
        """
        super().__init__(tool_id)
        self.detail = "Tool not found"

__init__ 

__init__(tool_id: UUID) -> None

Initialize tool not found error.

Parameters:

Name Type Description Default
tool_id UUID

The UUID of the tool.

 required
Source code in src/tool_inventory/connections.py 
51
52
53
54
55
56
57
58
def __init__(self, tool_id: UUID, /) -> None:
    """Initialize tool not found error.

    Args:
        tool_id: The UUID of the tool.
    """
    super().__init__(tool_id)
    self.detail = "Tool not found"

setup_database 

setup_database() -> None

Setup database.

Source code in src/tool_inventory/connections.py 
208
209
210
def setup_database() -> None:
    """Setup database."""
    SQLModel.metadata.create_all(engine)