#!/opt/gitlab/embedded/bin/python3
"""
When this module is installed as a file hook in GitLab, it forwards the event to a web service for further handling
"""
import logging
import logging.config
import requests
from configparser import ConfigParser
import sys
from enum import Enum
import os
import pathlib
from typing import Any, Callable, Dict, Iterator, List, Optional, Tuple
[docs]def logger_setup(filepath: str) -> Dict[str, Any]:
"""
Returns a dictionary which can be used to configure a logger.
:param filepath: path of the log file
:return: a dictionary to configure a logger
"""
return {
"version": 1,
"disable_existing_loggers": False,
"formatters": {
"standard": {
"format": "%(asctime)s %(levelname)-8s [%(filename)-20s:%(lineno)4d]: %(message)s"
},
},
"handlers": {
"file_handler": {
"level": "INFO",
"filename": filepath,
"class": "logging.handlers.RotatingFileHandler",
"formatter": "standard",
"maxBytes": 524288,
"backupCount": 2,
}
},
"loggers": {
"": {"handlers": ["file_handler"], "level": "INFO", "propagate": True},
},
}
[docs]def read_gitlab_event() -> str:
"""
Reads the GitLab system hook event from stdin.
:return: The event
"""
event_content = ""
for line in sys.stdin:
event_content = event_content + line
return event_content
[docs]class ConfigType(Enum):
"""
Enum for choosing the desired configuration
"""
PRODUCTION = 0
TEST = 1
DEBUG = 2
STAGING = 3
LOCAL = 4
[docs]def load_config(config_type = ConfigType.PRODUCTION):
filehooks_dir = pathlib.Path(os.path.join(os.path.dirname(__file__), '..', 'filehooks'))
config_file = "config.ini"
if config_type == ConfigType.PRODUCTION:
config_file = pathlib.Path("/file-hooks-src", "filehooks", "config.ini")
if config_type == ConfigType.STAGING:
config_file = pathlib.Path("/file-hooks-src", "filehooks", "config.ini")
if config_type == ConfigType.LOCAL:
config_file = pathlib.Path(filehooks_dir, "conf/localConfig.ini")
if config_type == ConfigType.DEBUG:
config_file = pathlib.Path(filehooks_dir, "conf/conf.debug.ini")
if config_type == ConfigType.TEST:
config_file = pathlib.Path(filehooks_dir, "conf/conf.test.ini")
config_parser = ConfigParser()
config_parser.read(config_file)
loaded_config = {s: dict(config_parser.items(s)) for s in config_parser.sections()}
return loaded_config
if __name__ == "__main__": # pragma: no cover
config = load_config()
_LOG_FILE = "/var/log/gitlab/gitlab-rails/trigger_project_update_local.log"
logger_setup = logger_setup(_LOG_FILE)
logging.config.dictConfig(logger_setup)
logger = logging.getLogger(__name__)
event = read_gitlab_event()
indexServiceURL = config["indexService"]["url"]
if event is not None:
headers = {'Content-Type': 'application/json; charset=utf-8'}
response = requests.post(indexServiceURL, headers=headers, data=event)
logger.info('forwarded event to %s: Status %s', indexServiceURL, response.status_code)