<commit_before>from __future__ import annotations

import atexit
import logging
import sys
import threading
from collections.abc import MutableMapping
from contextlib import ExitStack
from datetime import timedelta
from functools import partial
from logging import Logger
from types import TracebackType
from typing import Any, Callable, Iterable, Mapping
from uuid import UUID

from anyio import start_blocking_portal
from anyio.from_thread import BlockingPortal

from .. import Event, current_scheduler
from .._enums import CoalescePolicy, ConflictPolicy, RunState, SchedulerRole
from .._structures import JobResult, Schedule
from ..abc import DataStore, EventBroker, JobExecutor, Subscription, Trigger
from .async_ import AsyncScheduler

if sys.version_info >= (3, 11):
    from typing import Self
else:
    from typing_extensions import Self


class Scheduler:
    """A synchronous scheduler implementation."""

    def __init__(
        self,
        data_store: DataStore | None = None,
        event_broker: EventBroker | None = None,
        *,
        identity: str | None = None,
        role: SchedulerRole = SchedulerRole.both,
        job_executors: Mapping[str, JobExecutor] | None = None,
        default_job_executor: str | None = None,
        logger: Logger | None = None,
    ):
        kwargs: dict[str, Any] = {}
        if data_store is not None:
            kwargs["data_store"] = data_store
        if event_broker is not None:
            kwargs["event_broker"] = event_broker

        if not default_job_executor and not job_executors:
            default_job_executor = "threadpool"

        self._async_scheduler = AsyncScheduler(
            identity=identity,
            role=role,
            job_executors=job_executors,
            default_job_executor=default_job_executor,
            logger=logger or logging.getLogger(__name__),
            **kwargs,
        )
        self._exit_stack = ExitStack()
        self._portal: BlockingPortal | None = None
        self._lock = threading.RLock()

    @property
    def data_store(self) -> DataStore:
        return self._async_scheduler.data_store

    @property
    def event_broker(self) -> EventBroker:
        return self._async_scheduler.event_broker

    @property
    def identity(self) -> str:
        return self._async_scheduler.identity

    @property
    def role(self) -> SchedulerRole:
        return self._async_scheduler.role

    @property
    def job_executors(self) -> MutableMapping[str, JobExecutor]:
        return self._async_scheduler.job_executors

    @property
    def state(self) -> RunState:
        """The current running state of the scheduler."""
        return self._async_scheduler.state

    def __enter__(self: Self) -> Self:
        self._ensure_services_ready(self._exit_stack)
        return self

    def __exit__(
        self,
        exc_type: type[BaseException],
        exc_val: BaseException,
        exc_tb: TracebackType,<commit_msg>Added property for default_job_executor to SyncScheduler<commit_after>from __future__ import annotations

import atexit
import logging
import sys
import threading
from collections.abc import MutableMapping
from contextlib import ExitStack
from datetime import timedelta
from functools import partial
from logging import Logger
from types import TracebackType
from typing import Any, Callable, Iterable, Mapping
from uuid import UUID

from anyio import start_blocking_portal
from anyio.from_thread import BlockingPortal

from .. import Event, current_scheduler
from .._enums import CoalescePolicy, ConflictPolicy, RunState, SchedulerRole
from .._structures import JobResult, Schedule
from ..abc import DataStore, EventBroker, JobExecutor, Subscription, Trigger
from .async_ import AsyncScheduler

if sys.version_info >= (3, 11):
    from typing import Self
else:
    from typing_extensions import Self


class Scheduler:
    """A synchronous scheduler implementation."""

    def __init__(
        self,
        data_store: DataStore | None = None,
        event_broker: EventBroker | None = None,
        *,
        identity: str | None = None,
        role: SchedulerRole = SchedulerRole.both,
        job_executors: Mapping[str, JobExecutor] | None = None,
        default_job_executor: str | None = None,
        logger: Logger | None = None,
    ):
        kwargs: dict[str, Any] = {}
        if data_store is not None:
            kwargs["data_store"] = data_store
        if event_broker is not None:
            kwargs["event_broker"] = event_broker

        if not default_job_executor and not job_executors:
            default_job_executor = "threadpool"

        self._async_scheduler = AsyncScheduler(
            identity=identity,
            role=role,
            job_executors=job_executors,
            default_job_executor=default_job_executor,
            logger=logger or logging.getLogger(__name__),
            **kwargs,
        )
        self._exit_stack = ExitStack()
        self._portal: BlockingPortal | None = None
        self._lock = threading.RLock()

    @property
    def data_store(self) -> DataStore:
        return self._async_scheduler.data_store

    @property
    def event_broker(self) -> EventBroker:
        return self._async_scheduler.event_broker

    @property
    def identity(self) -> str:
        return self._async_scheduler.identity

    @property
    def role(self) -> SchedulerRole:
        return self._async_scheduler.role

    @property
    def job_executors(self) -> MutableMapping[str, JobExecutor]:
        return self._async_scheduler.job_executors

    @property
    def default_job_executor(self) -> str:
        return self._async_scheduler.default_job_executor

    @default_job_executor.setter
    def default_job_executor(self, value: str) -> None:
        self._async_scheduler.default_job_executor = value

    @property
    def state(self) -> RunState:
        """The current running state of the scheduler."""
        return self._async_scheduler.state

    def __enter__(self: Self) -> Self:
        self._ensure_services_ready(self._exit_stack)
        return self

    def __exit__(
        self,
        exc_type: type[BaseException],
        exc_val: BaseException,
        exc_tb: TracebackType,
{'fraction_non_alphanumeric': 0.07152020525978191, 'fraction_numerical': 0.0009621552277100705, 'mean_word_length': 2.9325346784363178, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 21, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3727330', 'n_tokens_mistral': 1763, 'n_tokens_neox': 1720, 'n_words': 490}