Skip to content

Services and schemas

Here is implemented the main functionality for API services.

Services

Main business logic implementation

Source code in app/services.py 
 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
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
class Services:
    """
    Main business logic implementation
    """

    def __init__(self):
        """
        Initiate preprocessor, validator, spacy validator and rule based validator
        """
        self.preprocessor = Preprocessor()
        self.validator = Validator()
        # self.spacy_validator = SpacyValidator(config.SPACY_MODEL_PATH)
        self.rule_based_validator = RuleBasedValidator()
        self.huggingface_validator = HuggingFaceValidator(config.HUGGINGFACE_MODEL_PATH,
                                                          config.HUGGINGFACE_PRETRAINED_MODEL)

    @staticmethod
    def post_process_predictions(sentences, predictions, probabilities, locations):
        """
        Post process model predictions

        Parameters
        ----------
        sentences : list
            The sentences of the remark
        predictions : list
            List of booleans each of which shows whether a sentence is valid or not
        probabilities : list
            Probabilities of each sentence prediction
        locations : list
            Position of each sentence in remark

        Returns
        -------
        sentences_prediction: list
            Prediction for each sentence
        is_all_valid: bool
            False if any of sentences in remark contain violation
        """

        is_all_valid = all(predictions)
        sentences_predictions = [SentencePredictions(is_valid=v, sentence=s, probability=p, location=l) for v, s, p, l
                                 in
                                 zip(predictions, sentences, probabilities, locations)]
        return sentences_predictions, is_all_valid

    def check_competitors(self, text, urduc):
        """
        Main logic for check_competitors endpoint

        Parameters
        ----------
        urduc : int
            enum for urduc
        text : str
            input string

        Returns
        -------
        pred : bool
            0 for violation
        violations : set[str]
            set of violations
        """

        if urduc not in SUPPORTED_URDUCS:
            raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=f'Not supported urduc:{urduc}')

        sentences, clean_sentences, locations = self.preprocessor.get_sentences(text)
        pred, violations = self.validator.validate_competitors_and_profanity(
            preprocessed_sentences=clean_sentences,
            text=text,
            urduc=urduc)

        # sentences_prediction, is_all_valid = self.post_process_predictions(sentences=sentences,
        #                                                                    predictions=predictions,
        #                                                                    probabilities=probabilities,
        #                                                                    )
        return pred, violations

    def validate_text(self, text, threshold, urduc):
        """
        Main logic for validate_text endpoint.
        Parameters
        ----------
        text : str
            Remark of a house
        threshold : float
            Threshold for violation detection

        Returns
        -------
        sentences_predictions : list
            Predictions for each sentence
        is_all_valid : bool
            False if any of sentences in remark contain violation
        violations: List[str]
            violations
        """

        if urduc not in SUPPORTED_URDUCS:
            raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=f'Not supported urduc:{urduc}')

        sentences, clean_sentences, locations = self.preprocessor.get_sentences(text, lowercase=False)

        bool_predictions, probabilities, violations = self.validator.validate(sentences=sentences,
                                                                              clean_sentences=clean_sentences,
                                                                              urduc=urduc, threshold=threshold)

        sentences_predictions, is_all_valid = self.post_process_predictions(sentences=sentences,
                                                                            predictions=bool_predictions,
                                                                            probabilities=probabilities,
                                                                            locations=locations)
        return sentences_predictions, is_all_valid, violations

    @staticmethod
    def post_process_hud_predictions(sentences, probabilities):
        """
        Post process HuggingFace model predictions
        Parameters
        ----------
        sentences : list
            The sentences of the remark
        probabilities : list
            Probabilities of each sentence prediction
        Returns
        -------
        sentences_predictions : list
            Prediction for each sentence
        """

        sentences_predictions = [SentenceViolationPredictions(sentence=sent, probability=prob) for sent, prob in
                                 zip(sentences, probabilities)]

        return sentences_predictions

    def validate_hud(self, text):
        """
        Main logic for validate_hud endpoint.
        Parameters
        ----------
        text : str
            Remark of a house

        Returns
        -------
        sentences_predictions : list
            Predictions for each sentence
        """
        if text == "":
            sentences = []
            probabilities = []
        else:
            sentences, clean_sentences, locations = self.preprocessor.get_sentences(text, lowercase=False)
            probabilities = self.huggingface_validator.validate(sentences=sentences)
        sentences_predictions = self.post_process_hud_predictions(sentences, probabilities)
        return sentences_predictions

__init__()

Initiate preprocessor, validator, spacy validator and rule based validator

Source code in app/services.py 
16
17
18
19
20
21
22
23
24
25
def __init__(self):
    """
    Initiate preprocessor, validator, spacy validator and rule based validator
    """
    self.preprocessor = Preprocessor()
    self.validator = Validator()
    # self.spacy_validator = SpacyValidator(config.SPACY_MODEL_PATH)
    self.rule_based_validator = RuleBasedValidator()
    self.huggingface_validator = HuggingFaceValidator(config.HUGGINGFACE_MODEL_PATH,
                                                      config.HUGGINGFACE_PRETRAINED_MODEL)

post_process_predictions(sentences, predictions, probabilities, locations) staticmethod

Post process model predictions

Parameters:

Name Type Description Default
sentences list

The sentences of the remark

 required
predictions list

List of booleans each of which shows whether a sentence is valid or not

 required
probabilities list

Probabilities of each sentence prediction

 required
locations list

Position of each sentence in remark

 required

Returns:

Name Type Description
sentences_prediction list

Prediction for each sentence
is_all_valid bool

False if any of sentences in remark contain violation
Source code in app/services.py 
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
@staticmethod
def post_process_predictions(sentences, predictions, probabilities, locations):
    """
    Post process model predictions

    Parameters
    ----------
    sentences : list
        The sentences of the remark
    predictions : list
        List of booleans each of which shows whether a sentence is valid or not
    probabilities : list
        Probabilities of each sentence prediction
    locations : list
        Position of each sentence in remark

    Returns
    -------
    sentences_prediction: list
        Prediction for each sentence
    is_all_valid: bool
        False if any of sentences in remark contain violation
    """

    is_all_valid = all(predictions)
    sentences_predictions = [SentencePredictions(is_valid=v, sentence=s, probability=p, location=l) for v, s, p, l
                             in
                             zip(predictions, sentences, probabilities, locations)]
    return sentences_predictions, is_all_valid

check_competitors(text, urduc)

Main logic for check_competitors endpoint

Parameters:

Name Type Description Default
urduc int

enum for urduc

 required
text str

input string

 required

Returns:

Name Type Description
pred bool

0 for violation
violations set[str]

set of violations
Source code in app/services.py 
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
def check_competitors(self, text, urduc):
    """
    Main logic for check_competitors endpoint

    Parameters
    ----------
    urduc : int
        enum for urduc
    text : str
        input string

    Returns
    -------
    pred : bool
        0 for violation
    violations : set[str]
        set of violations
    """

    if urduc not in SUPPORTED_URDUCS:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=f'Not supported urduc:{urduc}')

    sentences, clean_sentences, locations = self.preprocessor.get_sentences(text)
    pred, violations = self.validator.validate_competitors_and_profanity(
        preprocessed_sentences=clean_sentences,
        text=text,
        urduc=urduc)

    # sentences_prediction, is_all_valid = self.post_process_predictions(sentences=sentences,
    #                                                                    predictions=predictions,
    #                                                                    probabilities=probabilities,
    #                                                                    )
    return pred, violations

validate_text(text, threshold, urduc)

Main logic for validate_text endpoint.

Parameters:

Name Type Description Default
text str

Remark of a house

 required
threshold float

Threshold for violation detection

 required

Returns:

Name Type Description
sentences_predictions list

Predictions for each sentence
is_all_valid bool

False if any of sentences in remark contain violation
violations List[str]

violations
Source code in app/services.py 
 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
def validate_text(self, text, threshold, urduc):
    """
    Main logic for validate_text endpoint.
    Parameters
    ----------
    text : str
        Remark of a house
    threshold : float
        Threshold for violation detection

    Returns
    -------
    sentences_predictions : list
        Predictions for each sentence
    is_all_valid : bool
        False if any of sentences in remark contain violation
    violations: List[str]
        violations
    """

    if urduc not in SUPPORTED_URDUCS:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=f'Not supported urduc:{urduc}')

    sentences, clean_sentences, locations = self.preprocessor.get_sentences(text, lowercase=False)

    bool_predictions, probabilities, violations = self.validator.validate(sentences=sentences,
                                                                          clean_sentences=clean_sentences,
                                                                          urduc=urduc, threshold=threshold)

    sentences_predictions, is_all_valid = self.post_process_predictions(sentences=sentences,
                                                                        predictions=bool_predictions,
                                                                        probabilities=probabilities,
                                                                        locations=locations)
    return sentences_predictions, is_all_valid, violations

post_process_hud_predictions(sentences, probabilities) staticmethod

Post process HuggingFace model predictions

Parameters:

Name Type Description Default
sentences list

The sentences of the remark

 required
probabilities list

Probabilities of each sentence prediction

 required

Returns:

Name Type Description
sentences_predictions list

Prediction for each sentence
Source code in app/services.py 
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
@staticmethod
def post_process_hud_predictions(sentences, probabilities):
    """
    Post process HuggingFace model predictions
    Parameters
    ----------
    sentences : list
        The sentences of the remark
    probabilities : list
        Probabilities of each sentence prediction
    Returns
    -------
    sentences_predictions : list
        Prediction for each sentence
    """

    sentences_predictions = [SentenceViolationPredictions(sentence=sent, probability=prob) for sent, prob in
                             zip(sentences, probabilities)]

    return sentences_predictions

validate_hud(text)

Main logic for validate_hud endpoint.

Parameters:

Name Type Description Default
text str

Remark of a house

 required

Returns:

Name Type Description
sentences_predictions list

Predictions for each sentence
Source code in app/services.py 
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
def validate_hud(self, text):
    """
    Main logic for validate_hud endpoint.
    Parameters
    ----------
    text : str
        Remark of a house

    Returns
    -------
    sentences_predictions : list
        Predictions for each sentence
    """
    if text == "":
        sentences = []
        probabilities = []
    else:
        sentences, clean_sentences, locations = self.preprocessor.get_sentences(text, lowercase=False)
        probabilities = self.huggingface_validator.validate(sentences=sentences)
    sentences_predictions = self.post_process_hud_predictions(sentences, probabilities)
    return sentences_predictions

Schemas for endpoints

ValidateSentencesRequest

Bases: BaseModel

Request model for /validate_text endpoint

Source code in app/schemas.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
class ValidateSentencesRequest(BaseModel):
    """
    Request model for /validate_text endpoint
    """
    text: str
    urduc: Literal[tuple(ALL_URDUCS)]
    home_id: Optional[str]
    threshold: Optional[confloat(strict=True, ge=0.0, lt=1.0)] = config.PREDICTION_THRESHOLD

    # example for documentation
    class Config:
        """
        Additional configs for model
        """

        schema_extra = {
            'example': {
                "home_id": "456123",
                "urduc": 2,
                "text": "Kitchen appliances not included. "
                        "Engel & Volkers and Signature is one of the world’s leading service companies."
            }
        }

Config

Additional configs for model

Source code in app/schemas.py 
20
21
22
23
24
25
26
27
28
29
30
31
32
class Config:
    """
    Additional configs for model
    """

    schema_extra = {
        'example': {
            "home_id": "456123",
            "urduc": 2,
            "text": "Kitchen appliances not included. "
                    "Engel & Volkers and Signature is one of the world’s leading service companies."
        }
    }

SentenceLocation

Bases: BaseModel

Model for single sentence location

Source code in app/schemas.py 
35
36
37
38
39
40
class SentenceLocation(BaseModel):
    """
    Model for single sentence location
    """
    start: int
    end: int

SentencePredictions

Bases: BaseModel

Model for single sentence prediction

Source code in app/schemas.py 
43
44
45
46
47
48
49
50
class SentencePredictions(BaseModel):
    """
    Model for single sentence prediction
    """
    sentence: str
    is_valid: bool
    probability: float
    location: SentenceLocation

ValidateSentencesResponse

Bases: BaseModel

Response model for /validate_text endpoint

Source code in app/schemas.py 
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
class ValidateSentencesResponse(BaseModel):
    """
    Response model for /validate_text endpoint
    """
    sentences: Optional[List[SentencePredictions]]
    home_id: Optional[str]
    is_all_valid: bool
    banned_words: list

    # example for documentation
    class Config:
        """
        Additional configs for model
        """

        schema_extra = {'example': {
            "sentences": [
                {
                    "sentence": "Kitchen appliances not included.",
                    "is_valid": True,
                    "probability": 9.736195352161303e-05
                },
                {
                    "sentence": "Engel & Volkers and Signature is one of the world’s leading service companies.",
                    "is_valid": False,
                    "probability": 1.0
                }
            ],
            "home_id": "456123",
            "is_all_valid": False,
            "banned_words": ["Engel & Volkers",
                             "Signature"
                             ]
        }
        }

Config

Additional configs for model

Source code in app/schemas.py 
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
class Config:
    """
    Additional configs for model
    """

    schema_extra = {'example': {
        "sentences": [
            {
                "sentence": "Kitchen appliances not included.",
                "is_valid": True,
                "probability": 9.736195352161303e-05
            },
            {
                "sentence": "Engel & Volkers and Signature is one of the world’s leading service companies.",
                "is_valid": False,
                "probability": 1.0
            }
        ],
        "home_id": "456123",
        "is_all_valid": False,
        "banned_words": ["Engel & Volkers",
                         "Signature"
                         ]
    }
    }

ValidateCheckCompetitorsResponse

Bases: BaseModel

Response model for /validate_sentences endpoint

Source code in app/schemas.py 
 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
class ValidateCheckCompetitorsResponse(BaseModel):
    """
    Response model for /validate_sentences endpoint
    """
    sentences: Optional[List[SentencePredictions]]
    home_id: Optional[str]
    is_all_valid: bool
    banned_words: List[str] = []

    # example for documentation
    class Config:
        """
        Additional configs for model
        """

        schema_extra = {'example': {
            "sentences": [
                {
                    "sentence": "Kitchen appliances not included.",
                    "is_valid": True,
                    "probability": 9.736195352161303e-05
                },
                {
                    "sentence": "Engel & Volkers and Signature is one of the world’s leading service companies.",
                    "is_valid": False,
                    "probability": 1.0
                }
            ],
            "home_id": "456123",
            "is_all_valid": False,
            "banned_words": [
                "Engel & Volkers",
                "Signature"
            ]
        }}

Config

Additional configs for model

Source code in app/schemas.py 
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
class Config:
    """
    Additional configs for model
    """

    schema_extra = {'example': {
        "sentences": [
            {
                "sentence": "Kitchen appliances not included.",
                "is_valid": True,
                "probability": 9.736195352161303e-05
            },
            {
                "sentence": "Engel & Volkers and Signature is one of the world’s leading service companies.",
                "is_valid": False,
                "probability": 1.0
            }
        ],
        "home_id": "456123",
        "is_all_valid": False,
        "banned_words": [
            "Engel & Volkers",
            "Signature"
        ]
    }}

SentenceViolationPredictions

Bases: BaseModel

Model for single sentence fair housing violation prediction

Source code in app/schemas.py 
127
128
129
130
131
132
class SentenceViolationPredictions(BaseModel):
    """
    Model for single sentence fair housing violation prediction
    """
    sentence: str
    probability: float

ValidateHUDRequest

Bases: BaseModel

Request model for /validate_hud endpoint

Source code in app/schemas.py 
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
class ValidateHUDRequest(BaseModel):
    """
    Request model for /validate_hud endpoint
    """
    text: str
    home_id: Optional[str]

    # example for documentation
    class Config:
        """
        Additional configs for model
        """

        schema_extra = {
            'example': {
                "home_id": "456123",
                "text": "House needs TLC and is great for Jewish Americans with a vision.Enormous side and backyard, "
                        "perfect for gardening and entertaining!"
            }
        }

Config

Additional configs for model

Source code in app/schemas.py 
143
144
145
146
147
148
149
150
151
152
153
154
class Config:
    """
    Additional configs for model
    """

    schema_extra = {
        'example': {
            "home_id": "456123",
            "text": "House needs TLC and is great for Jewish Americans with a vision.Enormous side and backyard, "
                    "perfect for gardening and entertaining!"
        }
    }

ValidateHUDResponse

Bases: BaseModel

Response model for /validate_hud endpoint

Source code in app/schemas.py 
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
class ValidateHUDResponse(BaseModel):
    """
    Response model for /validate_hud endpoint
    """
    sentences: Optional[List[SentenceViolationPredictions]]
    home_id: Optional[str]

    # example for documentation
    class Config:
        """
        Additional configs for model
        """

        schema_extra = {'example': {
            "sentences": [
                {
                    "sentence": "House needs TLC and is great for Jewish Americans with a vision.",
                    "probability": 0.98840183
                },
                {
                    "sentence": "Enormous side and backyard, perfect for gardening and entertaining!",
                    "probability": 0.05458653
                }
            ],
            "home_id": "456123"
        }}

Config

Additional configs for model

Source code in app/schemas.py 
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
class Config:
    """
    Additional configs for model
    """

    schema_extra = {'example': {
        "sentences": [
            {
                "sentence": "House needs TLC and is great for Jewish Americans with a vision.",
                "probability": 0.98840183
            },
            {
                "sentence": "Enormous side and backyard, perfect for gardening and entertaining!",
                "probability": 0.05458653
            }
        ],
        "home_id": "456123"
    }}

Middlewares for pre- and post- processing of requests/responses

LoggingMiddleware

Bases: BaseHTTPMiddleware

Basic logging middleware inherited from starlette.BaseHTTPMiddleware

Source code in app/middlewares.py 
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
class LoggingMiddleware(BaseHTTPMiddleware):
    """
    Basic logging middleware inherited from starlette.BaseHTTPMiddleware
    """

    def __init__(self, app, logger):
        """
        Init object with app and logger
        Parameters
        ----------
        app : fast.FastAPI
            application object where middleware need to be added
        logger : logging.loger
            already configured logger for logging requests
        """
        super().__init__(app)
        self.logger = logger

    async def dispatch(self, request: Request, call_next):
        """
        Overriding BaseHTTPMiddleware.dispatch method to implement logging logic
        Parameters
        ----------
        request : starlette.middleware.base.Request
            current request
        call_next : starlette.middleware.base.RequestResponseEndpoint
            call function
        Returns
        -------
        streaming_response : starlette.middleware.base.StreamingResponse
            streaming response for the endpoint
        """
        guid = generate_guid()
        self.logger.info(f"rid={guid} start request path={request.url.path}")
        start_time = time()

        streaming_response = await call_next(request)
        status_code = streaming_response.status_code
        process_time = (time() - start_time) * 1000
        formatted_process_time = '{0:.2f}'.format(process_time)

        # collect errors and log also error messages
        if status_code != status.HTTP_200_OK:
            response_body = [
                section async for section in streaming_response.body_iterator
            ]
            streaming_response.body_iterator = iterate_in_threadpool(
                iter(response_body)
            )
            msg = response_body[0].decode()
            self.logger.info(
                f"rid={guid} completed_in={formatted_process_time}ms"
                f" status_code={status_code}"
                f" response = {response_body},"
                f" message = {msg}"
            )
        # for 200 OK requests logging only event
        else:
            self.logger.info(
                f"rid={guid} completed_in={formatted_process_time}ms status_code={status_code}"
            )

        return streaming_response

__init__(app, logger)

Init object with app and logger

Parameters:

Name Type Description Default
app FastAPI

application object where middleware need to be added

 required
logger loger

already configured logger for logging requests

 required
Source code in app/middlewares.py 
26
27
28
29
30
31
32
33
34
35
36
37
def __init__(self, app, logger):
    """
    Init object with app and logger
    Parameters
    ----------
    app : fast.FastAPI
        application object where middleware need to be added
    logger : logging.loger
        already configured logger for logging requests
    """
    super().__init__(app)
    self.logger = logger

dispatch(request, call_next) async

Overriding BaseHTTPMiddleware.dispatch method to implement logging logic

Parameters:

Name Type Description Default
request Request

current request

 required
call_next RequestResponseEndpoint

call function

 required

Returns:

Name Type Description
streaming_response StreamingResponse

streaming response for the endpoint
Source code in app/middlewares.py 
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
async def dispatch(self, request: Request, call_next):
    """
    Overriding BaseHTTPMiddleware.dispatch method to implement logging logic
    Parameters
    ----------
    request : starlette.middleware.base.Request
        current request
    call_next : starlette.middleware.base.RequestResponseEndpoint
        call function
    Returns
    -------
    streaming_response : starlette.middleware.base.StreamingResponse
        streaming response for the endpoint
    """
    guid = generate_guid()
    self.logger.info(f"rid={guid} start request path={request.url.path}")
    start_time = time()

    streaming_response = await call_next(request)
    status_code = streaming_response.status_code
    process_time = (time() - start_time) * 1000
    formatted_process_time = '{0:.2f}'.format(process_time)

    # collect errors and log also error messages
    if status_code != status.HTTP_200_OK:
        response_body = [
            section async for section in streaming_response.body_iterator
        ]
        streaming_response.body_iterator = iterate_in_threadpool(
            iter(response_body)
        )
        msg = response_body[0].decode()
        self.logger.info(
            f"rid={guid} completed_in={formatted_process_time}ms"
            f" status_code={status_code}"
            f" response = {response_body},"
            f" message = {msg}"
        )
    # for 200 OK requests logging only event
    else:
        self.logger.info(
            f"rid={guid} completed_in={formatted_process_time}ms status_code={status_code}"
        )

    return streaming_response

ExceptionHandlerMiddleware

Bases: BaseHTTPMiddleware

A middleware to handle errors

Source code in app/middlewares.py 
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
class ExceptionHandlerMiddleware(BaseHTTPMiddleware):
    """A middleware to handle errors"""

    async def dispatch(self, request: Request, call_next):
        """Try to process the request. If failed, return details about the exception"""

        try:
            return await call_next(request)
        except Exception as e:
            info = self.extract_info(e)
            status_code = 500
        content = {"detail": [info]}
        return JSONResponse(status_code=status_code, content=content)

    @staticmethod
    def extract_info(error: BaseException):
        """Extract the type and the message of an error and return as a dict"""
        return {"type": type(error).__name__,
                "msg": str(error)}

dispatch(request, call_next) async

Try to process the request. If failed, return details about the exception

Source code in app/middlewares.py 
89
90
91
92
93
94
95
96
97
98
async def dispatch(self, request: Request, call_next):
    """Try to process the request. If failed, return details about the exception"""

    try:
        return await call_next(request)
    except Exception as e:
        info = self.extract_info(e)
        status_code = 500
    content = {"detail": [info]}
    return JSONResponse(status_code=status_code, content=content)

extract_info(error) staticmethod

Extract the type and the message of an error and return as a dict

Source code in app/middlewares.py 
100
101
102
103
104
@staticmethod
def extract_info(error: BaseException):
    """Extract the type and the message of an error and return as a dict"""
    return {"type": type(error).__name__,
            "msg": str(error)}